diff --git a/docs/design/traj-exp-experience-learning-redesign.md b/docs/design/traj-exp-experience-learning-redesign.md
index fa8b179689..19637571bb 100644
--- a/docs/design/traj-exp-experience-learning-redesign.md
+++ b/docs/design/traj-exp-experience-learning-redesign.md
@@ -650,6 +650,7 @@ merge 输入/输出日志通过 `tracer.info(..., console=False)` 记录，避
"trace_id": "...",
"extracted_at": "...",
"operations": {...},
+ "skipped_operations": [...],
"summary": {...}
}
```
diff --git a/docs/en/api/05-sessions.md b/docs/en/api/05-sessions.md
index 2f54e7461d..f37271c599 100644
--- a/docs/en/api/05-sessions.md
+++ b/docs/en/api/05-sessions.md
@@ -1612,10 +1612,19 @@ When long-term memory extraction runs successfully, the commit writes a `memory_
}
]
},
+ "skipped_operations": [
+ {
+ "memory_type": "events",
+ "page_id": 101,
+ "reason_code": "invalid_ranges",
+ "reason": "No valid event range could be resolved"
+ }
+ ],
"summary": {
"total_adds": 1,
"total_updates": 1,
- "total_deletes": 1
+ "total_deletes": 1,
+ "total_skipped": 1
}
}
```
@@ -1627,11 +1636,13 @@ When long-term memory extraction runs successfully, the commit writes a `memory_
| `operations.adds` | array | New memories created (`uri`, `memory_type`, `after`) |
| `operations.updates` | array | Modified memories (`uri`, `memory_type`, `before`, `after`) |
| `operations.deletes` | array | Deleted memories (`uri`, `memory_type`, `deleted_content`) |
+| `skipped_operations` | array | Intentionally skipped operations and their stable reason codes; these do not represent file changes |
| `summary.total_adds` | int | Number of new memories |
| `summary.total_updates` | int | Number of modified memories |
| `summary.total_deletes` | int | Number of deleted memories |
+| `summary.total_skipped` | int | Number of intentionally skipped operations |
-An empty `memory_diff.json` (all counts zero) is written when long-term memory extraction runs but produces no memory operations.
+An empty `memory_diff.json` (all counts zero) is written when long-term memory extraction runs but produces no applied or intentionally skipped operations.
diff --git a/docs/en/concepts/08-session.md b/docs/en/concepts/08-session.md
index cdc7cd1d18..04f32074b3 100644
--- a/docs/en/concepts/08-session.md
+++ b/docs/en/concepts/08-session.md
@@ -205,10 +205,19 @@ Each `session.commit()` writes a `memory_diff.json` to the archive directory, re
}
]
},
+ "skipped_operations": [
+ {
+ "memory_type": "events",
+ "page_id": 101,
+ "reason_code": "invalid_ranges",
+ "reason": "No valid event range could be resolved"
+ }
+ ],
"summary": {
"total_adds": 1,
"total_updates": 1,
- "total_deletes": 1
+ "total_deletes": 1,
+ "total_skipped": 1
}
}
```
@@ -220,9 +229,10 @@ Each `session.commit()` writes a `memory_diff.json` to the archive directory, re
| `operations.adds` | New memories created (no `before`) |
| `operations.updates` | Modified memories (with `before` and `after`) |
| `operations.deletes` | Deleted memories (with `deleted_content`) |
+| `skipped_operations` | Intentionally skipped operations and their stable reason codes; these are not file changes |
| `summary` | Counts per operation type |
-An empty `memory_diff.json` (all counts zero) is written even when no memory operations occurred.
+An empty `memory_diff.json` (all counts zero) is written when no applied or intentionally skipped operations occurred.
## Storage Structure
diff --git a/docs/zh/api/05-sessions.md b/docs/zh/api/05-sessions.md
index c78470ae6e..5a6fcd38d8 100644
--- a/docs/zh/api/05-sessions.md
+++ b/docs/zh/api/05-sessions.md
@@ -1584,10 +1584,19 @@ viking://user/{user_id}/sessions/{session_id}/
}
]
},
+ "skipped_operations": [
+ {
+ "memory_type": "events",
+ "page_id": 101,
+ "reason_code": "invalid_ranges",
+ "reason": "无法解析出有效的事件范围"
+ }
+ ],
"summary": {
"total_adds": 1,
"total_updates": 1,
- "total_deletes": 1
+ "total_deletes": 1,
+ "total_skipped": 1
}
}
```
@@ -1599,11 +1608,13 @@ viking://user/{user_id}/sessions/{session_id}/
| `operations.adds` | array | 新增记忆（`uri`、`memory_type`、`after`） |
| `operations.updates` | array | 修改记忆（`uri`、`memory_type`、`before`、`after`） |
| `operations.deletes` | array | 删除记忆（`uri`、`memory_type`、`deleted_content`） |
+| `skipped_operations` | array | 策略性跳过的操作及稳定原因码；不代表文件变更 |
| `summary.total_adds` | int | 新增记忆数 |
| `summary.total_updates` | int | 修改记忆数 |
| `summary.total_deletes` | int | 删除记忆数 |
+| `summary.total_skipped` | int | 策略性跳过的操作数 |
-如果长记忆抽取已运行但没有产生记忆操作，也会写入空结构的 `memory_diff.json`（所有计数为零）。
+如果长记忆抽取已运行但没有产生实际变更或策略性跳过，也会写入空结构的 `memory_diff.json`（所有计数为零）。
diff --git a/docs/zh/concepts/08-session.md b/docs/zh/concepts/08-session.md
index 052684b019..d990d5acbd 100644
--- a/docs/zh/concepts/08-session.md
+++ b/docs/zh/concepts/08-session.md
@@ -205,10 +205,19 @@ LLM 去重决策 → candidate(skip/create/none) + item(merge/delete)
}
]
},
+ "skipped_operations": [
+ {
+ "memory_type": "events",
+ "page_id": 101,
+ "reason_code": "invalid_ranges",
+ "reason": "无法解析出有效的事件范围"
+ }
+ ],
"summary": {
"total_adds": 1,
"total_updates": 1,
- "total_deletes": 1
+ "total_deletes": 1,
+ "total_skipped": 1
}
}
```
@@ -220,9 +229,10 @@ LLM 去重决策 → candidate(skip/create/none) + item(merge/delete)
| `operations.adds` | 新增的记忆（无 `before`） |
| `operations.updates` | 修改的记忆（含 `before` 和 `after`） |
| `operations.deletes` | 删除的记忆（含 `deleted_content`） |
+| `skipped_operations` | 策略性跳过的操作及稳定原因码；不代表文件变更 |
| `summary` | 各操作类型的计数 |
-即使没有记忆操作，也会写入空结构的 `memory_diff.json`（所有计数为零）。
+如果没有实际变更或策略性跳过，也会写入空结构的 `memory_diff.json`（所有计数为零）。
## 存储结构
diff --git a/openviking/session/compressor_v3.py b/openviking/session/compressor_v3.py
index c568d3b86f..4be9145edc 100644
--- a/openviking/session/compressor_v3.py
+++ b/openviking/session/compressor_v3.py
@@ -346,6 +346,9 @@ async def _build_memory_diff(
adds=adds,
updates=updates,
deletes=deletes,
+ skipped_operations=_serialize_skipped_operations(
+ getattr(result, "skipped_operations", [])
+ ),
)
@tracer(ignore_result=True)
@@ -363,6 +366,7 @@ async def extract_long_term_memories(
allow_self_memory: bool = True,
allowed_peer_ids: Optional[set[str]] = None,
event_search_tags: Optional[List[str]] = None,
+ peer_memory_enabled: bool = True,
):
if not agent_evolution_enabled:
effective_types = (
@@ -400,6 +404,7 @@ async def extract_long_term_memories(
archive_uri=archive_uri,
allowed_memory_types=allowed_memory_types,
allow_self_memory=allow_self_memory,
+ peer_memory_enabled=peer_memory_enabled,
allowed_peer_ids=allowed_peer_ids,
event_search_tags=event_search_tags,
)
@@ -453,6 +458,7 @@ async def extract_long_term_memories(
contexts=result.contexts,
train_result=train_result,
archive_uri=archive_uri or "",
+ skipped_operations=getattr(result, "skipped_operations", []),
)
except Exception:
if strict_extract_errors:
@@ -601,6 +607,7 @@ async def _extract_user_memories(
archive_uri: Optional[str] = None,
allowed_memory_types: Optional[set[str]] = None,
allow_self_memory: bool = True,
+ peer_memory_enabled: bool = True,
allowed_peer_ids: Optional[set[str]] = None,
event_search_tags: Optional[List[str]] = None,
) -> "_V3ExtractionResult":
@@ -642,6 +649,7 @@ async def _extract_user_memories(
allowed_memory_types=allowed_memory_types,
allow_self=allow_self_memory,
allowed_peer_ids=allowed_peer_ids,
+ peer_memory_enabled=peer_memory_enabled,
)
isolation_handler.prepare_messages()
context_provider._isolation_handler = isolation_handler
@@ -682,6 +690,7 @@ async def _extract_user_memories(
"allowed_memory_types": allowed_memory_types,
"allow_self": allow_self_memory,
"allowed_peer_ids": allowed_peer_ids,
+ "peer_memory_enabled": peer_memory_enabled,
},
metadata={
"source_extraction_id": extraction_id,
@@ -719,6 +728,9 @@ async def _extract_user_memories(
cases=canonical_cases,
memory_diff=memory_diff,
case_uri_by_name=_case_uri_by_name(canonical_cases, patch_operations, result),
+ skipped_operations=_serialize_skipped_operations(
+ getattr(result, "skipped_operations", [])
+ ),
)
def _session_skill_extraction_enabled(self) -> bool:
@@ -1223,6 +1235,7 @@ class _V3ExtractionResult:
cases: list[Case] = field(default_factory=list)
memory_diff: dict[str, Any] | None = None
case_uri_by_name: dict[str, str] = field(default_factory=dict)
+ skipped_operations: list[dict[str, Any]] = field(default_factory=list)
@dataclass(slots=True)
@@ -2041,20 +2054,36 @@ def _same_memory_file(before: Optional[MemoryFile], after: Optional[MemoryFile])
)
+def _serialize_skipped_operations(items: Any) -> list[dict[str, Any]]:
+ serialized: list[dict[str, Any]] = []
+ for item in list(items or []):
+ if isinstance(item, dict):
+ payload = dict(item)
+ else:
+ model_dump = getattr(item, "model_dump", None)
+ if not callable(model_dump):
+ continue
+ payload = model_dump(mode="json", exclude_none=True)
+ if isinstance(payload, dict):
+ payload.pop("source", None)
+ serialized.append(payload)
+ return serialized
+
+
def _v3_extraction_response(
*,
contexts: list[Context],
train_result: Any,
archive_uri: str,
+ skipped_operations: Optional[list[dict[str, Any]]] = None,
) -> list[Context] | dict[str, Any]:
"""Build the extraction response.
Historically ``extract_long_term_memories`` returned ``list[Context]`` and
a number of direct callers still index/compare the return value as a list.
- Commit orchestration now also understands the execution-memory style
- ``{"contexts": ..., "session_skills": ...}`` shape so it can count
- session skills. Preserve the old list shape unless there are actual
- session skills to report.
+ Commit orchestration also understands a structured response for session
+ skills and intentionally skipped operations. Preserve the old list shape
+ unless either field has content.
"""
skill_dicts: list[dict[str, Any]] = []
seen: set[str] = set()
@@ -2064,9 +2093,13 @@ def _v3_extraction_response(
if uri_str and uri_str not in seen:
seen.add(uri_str)
skill_dicts.append({"uri": uri_str, "archive_uri": archive_uri})
- if not skill_dicts:
+ public_skips = list(skipped_operations or [])
+ if not skill_dicts and not public_skips:
return contexts
- return {"contexts": contexts, "session_skills": skill_dicts}
+ response: dict[str, Any] = {"contexts": contexts, "session_skills": skill_dicts}
+ if public_skips:
+ response["skipped_operations"] = public_skips
+ return response
def _make_memory_diff(
@@ -2075,7 +2108,9 @@ def _make_memory_diff(
adds: list[dict[str, Any]],
updates: list[dict[str, Any]],
deletes: list[dict[str, Any]],
+ skipped_operations: Optional[list[dict[str, Any]]] = None,
) -> dict[str, Any]:
+ skipped = list(skipped_operations or [])
return {
"archive_uri": archive_uri,
"trace_id": tracer.get_trace_id() or None,
@@ -2085,10 +2120,12 @@ def _make_memory_diff(
"updates": list(updates),
"deletes": list(deletes),
},
+ "skipped_operations": skipped,
"summary": {
"total_adds": len(adds),
"total_updates": len(updates),
"total_deletes": len(deletes),
+ "total_skipped": len(skipped),
},
}
@@ -2101,12 +2138,16 @@ def _merge_memory_diffs(
adds: list[dict[str, Any]] = []
updates: list[dict[str, Any]] = []
deletes: list[dict[str, Any]] = []
+ skipped_operations: list[dict[str, Any]] = []
trace_id = tracer.get_trace_id() or None
for diff in diffs:
if not isinstance(diff, dict):
continue
if trace_id is None and diff.get("trace_id"):
trace_id = str(diff.get("trace_id"))
+ skipped_operations.extend(
+ item for item in diff.get("skipped_operations", []) if isinstance(item, dict)
+ )
operations = diff.get("operations")
if not isinstance(operations, dict):
continue
@@ -2118,6 +2159,7 @@ def _merge_memory_diffs(
adds=adds,
updates=updates,
deletes=deletes,
+ skipped_operations=skipped_operations,
)
merged["trace_id"] = trace_id
return merged
@@ -2130,7 +2172,8 @@ def _memory_diff_has_changes(diff: Any) -> bool:
if not isinstance(summary, dict):
return False
return any(
- int(summary.get(key) or 0) > 0 for key in ("total_adds", "total_updates", "total_deletes")
+ int(summary.get(key) or 0) > 0
+ for key in ("total_adds", "total_updates", "total_deletes", "total_skipped")
)
diff --git a/openviking/session/memory/dataclass.py b/openviking/session/memory/dataclass.py
index 0d06ef6e40..04bc77d23c 100644
--- a/openviking/session/memory/dataclass.py
+++ b/openviking/session/memory/dataclass.py
@@ -169,6 +169,40 @@ class MemoryOperationSource(BaseModel):
extracted_at: Optional[str] = None
+class MemoryOperationSkipCode(str, Enum):
+ """Stable reason codes for intentionally skipped memory operations."""
+
+ MEMORY_TYPE_FILTERED = "memory_type_filtered"
+ SELF_MEMORY_DISABLED = "self_memory_disabled"
+ PEER_MEMORY_DISABLED = "peer_memory_disabled"
+ INVALID_PEER_ID = "invalid_peer_id"
+ PEER_NOT_ALLOWED = "peer_not_allowed"
+ INVALID_RANGES = "invalid_ranges"
+ AMBIGUOUS_TARGET = "ambiguous_target"
+ NO_WRITABLE_TARGET = "no_writable_target"
+ DEPENDENT_DELETE_SUPPRESSED = "dependent_delete_suppressed"
+
+
+class MemoryOperationSkip(BaseModel):
+ """Internal policy/validation decision explaining why no URI was produced."""
+
+ reason_code: MemoryOperationSkipCode
+ reason: str
+
+
+class SkippedMemoryOperation(BaseModel):
+ """Structured, task-visible record for one intentionally skipped operation."""
+
+ memory_type: str
+ page_id: Optional[int] = None
+ uri: Optional[str] = None
+ reason_code: MemoryOperationSkipCode
+ reason: str
+ # Source is used only to scope shared streaming-batch results back to the
+ # submitting commit. It must never be serialized into the public task result.
+ source: Optional[MemoryOperationSource] = Field(default=None, exclude=True)
+
+
# ============================================================================
# Memory Field and Schema Definitions
# ============================================================================
@@ -298,6 +332,9 @@ class ResolvedOperation(BaseModel):
uris: List[str]
page_id: Optional[int] = None # Temporary page_id for link resolution (not persisted)
source: Optional[MemoryOperationSource] = None
+ # Runtime-only resolution decision. It is deliberately excluded from model
+ # serialization so it cannot enter later LLM merge prompts or memory files.
+ resolution_skip: Optional[MemoryOperationSkip] = Field(default=None, exclude=True)
# Custom scalar tags (already normalized as "key=value") to attach to this
# operation's memories in the vector index. None means "no tags"; used by
# event-memory auto-tagging. Not persisted in the memory file content.
diff --git a/openviking/session/memory/memory_isolation_handler.py b/openviking/session/memory/memory_isolation_handler.py
index d7ffd23923..b8d997f45b 100644
--- a/openviking/session/memory/memory_isolation_handler.py
+++ b/openviking/session/memory/memory_isolation_handler.py
@@ -7,7 +7,12 @@
from openviking.core.peer_id import safe_peer_id
from openviking.server.identity import RequestContext
-from openviking.session.memory.dataclass import MemoryTypeSchema, ResolvedOperation
+from openviking.session.memory.dataclass import (
+ MemoryOperationSkip,
+ MemoryOperationSkipCode,
+ MemoryTypeSchema,
+ ResolvedOperation,
+)
from openviking.session.memory.memory_updater import ExtractContext
from openviking.session.memory.utils.uri import generate_uri, render_template
from openviking_cli.utils import get_logger
@@ -16,6 +21,30 @@
_INTERNAL_MEMORY_TYPES = {"session_skills"}
_SELF_PEER_ID = "__self"
+_INVALID_PEER_MARKER = "_memory_resolution_invalid_peer_id"
+
+_SKIP_REASON_MESSAGES = {
+ MemoryOperationSkipCode.MEMORY_TYPE_FILTERED: (
+ "Memory type is outside the allowed extraction scope"
+ ),
+ MemoryOperationSkipCode.SELF_MEMORY_DISABLED: "Self memory writes are disabled",
+ MemoryOperationSkipCode.PEER_MEMORY_DISABLED: "Peer memory writes are disabled",
+ MemoryOperationSkipCode.INVALID_PEER_ID: "Target peer ID is invalid",
+ MemoryOperationSkipCode.PEER_NOT_ALLOWED: ("Target peer is outside the allowed memory scope"),
+ MemoryOperationSkipCode.INVALID_RANGES: "Message ranges are malformed or out of bounds",
+ MemoryOperationSkipCode.AMBIGUOUS_TARGET: "Memory ownership cannot be resolved uniquely",
+ MemoryOperationSkipCode.NO_WRITABLE_TARGET: "No writable memory target could be resolved",
+}
+
+_SKIP_REASON_PRIORITY = {
+ MemoryOperationSkipCode.INVALID_PEER_ID: 0,
+ MemoryOperationSkipCode.INVALID_RANGES: 1,
+ MemoryOperationSkipCode.PEER_MEMORY_DISABLED: 2,
+ MemoryOperationSkipCode.PEER_NOT_ALLOWED: 3,
+ MemoryOperationSkipCode.SELF_MEMORY_DISABLED: 4,
+ MemoryOperationSkipCode.AMBIGUOUS_TARGET: 5,
+ MemoryOperationSkipCode.NO_WRITABLE_TARGET: 6,
+}
@dataclass
@@ -26,6 +55,12 @@ class RoleScope:
peer_ids: List[str] = field(default_factory=list)
+@dataclass(frozen=True)
+class _TargetResolution:
+ target_ids: List[str] = field(default_factory=list)
+ skip_code: Optional[MemoryOperationSkipCode] = None
+
+
def peer_user_space(user_space: str, peer_id: str) -> str:
"""Return the user-space fragment for memory about a stable peer."""
if peer_id == _SELF_PEER_ID:
@@ -43,6 +78,7 @@ def __init__(
allowed_memory_types: Optional[Set[str]] = None,
allow_self: bool = True,
allowed_peer_ids: Optional[Set[str]] = None,
+ peer_memory_enabled: Optional[bool] = None,
):
self.ctx = ctx
self._extract_context = extract_context
@@ -58,6 +94,9 @@ def __init__(
}
self.allow_self = bool(allow_self)
self.allowed_peer_ids = peer_ids
+ self.peer_memory_enabled = (
+ bool(peer_memory_enabled) if peer_memory_enabled is not None else True
+ )
self.allow_peer = bool(peer_ids)
def prepare_messages(self) -> None:
@@ -69,15 +108,29 @@ def _messages(self) -> List[Any]:
return messages if isinstance(messages, list) else []
def _message_target_id(self, msg: Any) -> Optional[str]:
+ resolution = self._message_target_resolution(msg)
+ return resolution.target_ids[0] if resolution.target_ids else None
+
+ def _message_target_resolution(self, msg: Any) -> _TargetResolution:
if not self._is_peer_owner_message(msg):
- return None
+ return _TargetResolution()
raw_peer_id = getattr(msg, "peer_id", None)
+ if raw_peer_id in (None, ""):
+ if self.allow_self:
+ return _TargetResolution([_SELF_PEER_ID])
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.SELF_MEMORY_DISABLED)
peer_id = safe_peer_id(raw_peer_id)
- if peer_id and self._can_write_peer(peer_id):
- return peer_id
- if raw_peer_id in (None, "") and self.allow_self:
- return _SELF_PEER_ID
- return None
+ if not peer_id:
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.INVALID_PEER_ID)
+ if peer_id == _SELF_PEER_ID:
+ if self.allow_self:
+ return _TargetResolution([_SELF_PEER_ID])
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.SELF_MEMORY_DISABLED)
+ if not self.peer_memory_enabled:
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.PEER_MEMORY_DISABLED)
+ if not self._can_write_peer(peer_id):
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.PEER_NOT_ALLOWED)
+ return _TargetResolution([peer_id])
@staticmethod
def _is_peer_owner_message(msg: Any) -> bool:
@@ -115,11 +168,14 @@ def fill_identity_fields(
item_dict.pop("peer_id", None)
return
- peer_id = safe_peer_id(item_dict.get("peer_id"))
+ raw_peer_id = item_dict.get("peer_id")
+ peer_id = safe_peer_id(raw_peer_id)
if peer_id and peer_id != _SELF_PEER_ID:
item_dict["peer_id"] = peer_id
else:
item_dict.pop("peer_id", None)
+ if raw_peer_id not in (None, "", _SELF_PEER_ID):
+ item_dict[_INVALID_PEER_MARKER] = True
def allows_schema(self, memory_type_schema: MemoryTypeSchema) -> bool:
memory_type = getattr(memory_type_schema, "memory_type", "")
@@ -134,23 +190,11 @@ def allows_schema(self, memory_type_schema: MemoryTypeSchema) -> bool:
def _can_write_peer(self, peer_id: str) -> bool:
return self.allow_peer and peer_id in self.allowed_peer_ids
- def _unique_peer_target_id_in_messages(self) -> Optional[str]:
- targets = [
- peer_id
- for msg in self._messages()
- if (peer_id := safe_peer_id(getattr(msg, "peer_id", None)))
- and self._is_peer_owner_message(msg)
- and self._can_write_peer(peer_id)
- ]
- peer_ids = list(dict.fromkeys(targets))
- return peer_ids[0] if len(peer_ids) == 1 else None
-
- def _unique_target_id_in_messages(self) -> Optional[str]:
+ def _target_ids_in_messages(self) -> List[str]:
targets = [
target_id for msg in self._messages() if (target_id := self._message_target_id(msg))
]
- target_ids = list(dict.fromkeys(targets))
- return target_ids[0] if len(target_ids) == 1 else None
+ return list(dict.fromkeys(targets))
def _range_is_fully_in_bounds(self, ranges: Any) -> bool:
parts = str(ranges).split(",")
@@ -197,41 +241,93 @@ def render_schema_directories(self, memory_type_schema: MemoryTypeSchema) -> Lis
)
return directories
- def _range_targets(self, ranges: Any) -> tuple[List[str], bool]:
+ @staticmethod
+ def _preferred_skip_code(
+ skip_codes: List[MemoryOperationSkipCode],
+ ) -> MemoryOperationSkipCode:
+ if not skip_codes:
+ return MemoryOperationSkipCode.NO_WRITABLE_TARGET
+ return min(skip_codes, key=lambda code: _SKIP_REASON_PRIORITY.get(code, 999))
+
+ def _range_targets(self, ranges: Any) -> _TargetResolution:
if not ranges or not self._extract_context:
- return [], False
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.INVALID_RANGES)
range_is_fully_in_bounds = self._range_is_fully_in_bounds(ranges)
+ if not range_is_fully_in_bounds:
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.INVALID_RANGES)
try:
msg_range = self._extract_context.read_message_ranges(str(ranges))
except Exception:
logger.warning("Failed to parse memory ranges for peer memory: %s", ranges)
- return [], False
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.INVALID_RANGES)
target_ids = []
has_message = False
has_user_message = False
+ skip_codes: List[MemoryOperationSkipCode] = []
for msg_group in getattr(msg_range, "elements", []) or []:
for msg in msg_group:
has_message = True
if self._is_peer_owner_message(msg):
has_user_message = True
- target_id = self._message_target_id(msg)
- if target_id:
- target_ids.append(target_id)
- can_fallback = range_is_fully_in_bounds and has_message and not has_user_message
- return list(dict.fromkeys(target_ids)), can_fallback
-
- def _resolve_operation_target_id(self, raw_peer_id: Any) -> Optional[str]:
- peer_id = safe_peer_id(raw_peer_id)
- if peer_id == _SELF_PEER_ID and self.allow_self:
- return _SELF_PEER_ID
- if peer_id and self._can_write_peer(peer_id):
- return peer_id
+ resolution = self._message_target_resolution(msg)
+ target_ids.extend(resolution.target_ids)
+ if resolution.skip_code is not None:
+ skip_codes.append(resolution.skip_code)
+ target_ids = list(dict.fromkeys(target_ids))
+ if target_ids:
+ return _TargetResolution(target_ids)
+ if has_user_message:
+ return _TargetResolution(skip_code=self._preferred_skip_code(skip_codes))
+ if has_message:
+ fallback_targets = self._target_ids_in_messages()
+ if len(fallback_targets) == 1:
+ return _TargetResolution(fallback_targets)
+ if len(fallback_targets) > 1:
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.AMBIGUOUS_TARGET)
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.NO_WRITABLE_TARGET)
+
+ def _resolve_operation_target(self, raw_peer_id: Any) -> _TargetResolution:
if raw_peer_id not in (None, ""):
- return None
+ peer_id = safe_peer_id(raw_peer_id)
+ if not peer_id:
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.INVALID_PEER_ID)
+ if peer_id == _SELF_PEER_ID:
+ if self.allow_self:
+ return _TargetResolution([_SELF_PEER_ID])
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.SELF_MEMORY_DISABLED)
+ if not self.peer_memory_enabled:
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.PEER_MEMORY_DISABLED)
+ if not self._can_write_peer(peer_id):
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.PEER_NOT_ALLOWED)
+ return _TargetResolution([peer_id])
if self.allow_self:
- return _SELF_PEER_ID
- return self._unique_peer_target_id_in_messages()
+ return _TargetResolution([_SELF_PEER_ID])
+ peer_targets = list(
+ dict.fromkeys(
+ target
+ for msg in self._messages()
+ for target in self._message_target_resolution(msg).target_ids
+ if target != _SELF_PEER_ID
+ )
+ )
+ if len(peer_targets) == 1:
+ return _TargetResolution(peer_targets)
+ if len(peer_targets) > 1:
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.AMBIGUOUS_TARGET)
+ return _TargetResolution(skip_code=MemoryOperationSkipCode.NO_WRITABLE_TARGET)
+
+ @staticmethod
+ def _skip_operation(
+ operation: ResolvedOperation,
+ reason_code: MemoryOperationSkipCode,
+ ) -> List[str]:
+ operation.memory_fields.pop(_INVALID_PEER_MARKER, None)
+ operation.resolution_skip = MemoryOperationSkip(
+ reason_code=reason_code,
+ reason=_SKIP_REASON_MESSAGES[reason_code],
+ )
+ return []
def calculate_memory_uris(
self,
@@ -239,8 +335,14 @@ def calculate_memory_uris(
operation: ResolvedOperation,
extract_context: ExtractContext,
):
+ operation.resolution_skip = None
if not self.allows_schema(memory_type_schema):
- return []
+ reason_code = (
+ MemoryOperationSkipCode.SELF_MEMORY_DISABLED
+ if not self.allow_self and not getattr(memory_type_schema, "peer_enabled", True)
+ else MemoryOperationSkipCode.MEMORY_TYPE_FILTERED
+ )
+ return self._skip_operation(operation, reason_code)
if not self.ctx or not self.ctx.user:
return []
@@ -249,25 +351,31 @@ def calculate_memory_uris(
operation.memory_fields["user_id"] = user_id
target_ids: List[str] = []
+ skip_code: Optional[MemoryOperationSkipCode] = None
has_ranges = operation.memory_fields.get("ranges") is not None
if not getattr(memory_type_schema, "peer_enabled", True):
operation.memory_fields.pop("peer_id", None)
- target_ids = [_SELF_PEER_ID] if self.allow_self else []
+ if self.allow_self:
+ target_ids = [_SELF_PEER_ID]
+ else:
+ skip_code = MemoryOperationSkipCode.SELF_MEMORY_DISABLED
elif operation.memory_fields.get("ranges") is not None:
- target_ids, can_fallback = self._range_targets(
+ resolution = self._range_targets(
operation.memory_fields.get("ranges"),
)
- if not target_ids and can_fallback:
- fallback_target = self._unique_target_id_in_messages()
- if fallback_target:
- target_ids = [fallback_target]
+ target_ids = resolution.target_ids
+ skip_code = resolution.skip_code
operation.memory_fields.pop("peer_id", None)
else:
- target_id = self._resolve_operation_target_id(
- operation.memory_fields.get("peer_id"),
+ invalid_peer_id = bool(operation.memory_fields.pop(_INVALID_PEER_MARKER, False))
+ resolution = (
+ _TargetResolution(skip_code=MemoryOperationSkipCode.INVALID_PEER_ID)
+ if invalid_peer_id
+ else self._resolve_operation_target(operation.memory_fields.get("peer_id"))
)
- if target_id:
- target_ids = [target_id]
+ target_ids = resolution.target_ids
+ skip_code = resolution.skip_code
+ target_id = target_ids[0] if len(target_ids) == 1 else None
if target_id == _SELF_PEER_ID:
operation.memory_fields.pop("peer_id", None)
elif target_id:
@@ -276,7 +384,10 @@ def calculate_memory_uris(
operation.memory_fields.pop("peer_id", None)
if not target_ids:
- return []
+ return self._skip_operation(
+ operation,
+ skip_code or MemoryOperationSkipCode.NO_WRITABLE_TARGET,
+ )
# 文件
uris = set()
diff --git a/openviking/session/memory/memory_updater.py b/openviking/session/memory/memory_updater.py
index 40a7c5b16a..d1b6f23e7e 100644
--- a/openviking/session/memory/memory_updater.py
+++ b/openviking/session/memory/memory_updater.py
@@ -22,8 +22,10 @@
from openviking.server.identity import RequestContext
from openviking.session.memory.dataclass import (
MemoryFile,
+ MemoryOperationSkipCode,
ResolvedOperation,
ResolvedOperations,
+ SkippedMemoryOperation,
StoredLink,
)
from openviking.session.memory.memory_type_registry import MemoryTypeRegistry
@@ -711,6 +713,7 @@ def __init__(self):
self.written_uris: List[str] = []
self.edited_uris: List[str] = []
self.deleted_uris: List[str] = []
+ self.skipped_operations: List[SkippedMemoryOperation] = []
self.errors: List[Tuple[str, Exception]] = []
def add_written(self, uri: str) -> None:
@@ -725,11 +728,15 @@ def add_deleted(self, uri: str) -> None:
def add_error(self, uri: str, error: Exception) -> None:
self.errors.append((uri, error))
+ def add_skipped(self, operation: SkippedMemoryOperation) -> None:
+ self.skipped_operations.append(operation)
+
def summary(self) -> str:
return (
f"Written: {len(self.written_uris)}, "
f"Edited: {len(self.edited_uris)}, "
f"Deleted: {len(self.deleted_uris)}, "
+ f"Skipped: {len(self.skipped_operations)}, "
f"Errors: {len(self.errors)}"
)
@@ -873,12 +880,38 @@ async def apply_operations(
applicable_upserts: List[ResolvedOperation] = []
has_unresolved_upserts = False
+ has_unexplained_unresolved_upserts = False
for resolved_op in operations.upsert_operations:
if resolved_op.uris:
applicable_upserts.append(resolved_op)
continue
has_unresolved_upserts = True
error_target = f"{resolved_op.memory_type}(page_id={resolved_op.page_id})"
+ resolution_skip = getattr(resolved_op, "resolution_skip", None)
+ if resolution_skip is not None:
+ skipped = SkippedMemoryOperation(
+ memory_type=resolved_op.memory_type,
+ page_id=resolved_op.page_id,
+ reason_code=resolution_skip.reason_code,
+ reason=resolution_skip.reason,
+ source=resolved_op.source,
+ )
+ result.add_skipped(skipped)
+ message = (
+ "Skipping memory operation by resolution policy: "
+ f"memory_type={resolved_op.memory_type} "
+ f"page_id={resolved_op.page_id} "
+ f"reason_code={resolution_skip.reason_code.value}"
+ )
+ if resolution_skip.reason_code in {
+ MemoryOperationSkipCode.INVALID_PEER_ID,
+ MemoryOperationSkipCode.INVALID_RANGES,
+ }:
+ logger.warning(message)
+ else:
+ tracer.info(message)
+ continue
+ has_unexplained_unresolved_upserts = True
resolution_error = ValueError("Missing resolved URI")
result.add_error(error_target, resolution_error)
tracer.error(
@@ -932,6 +965,28 @@ async def apply_operations(
for file_content in operations.delete_file_contents:
delete_uri = file_content.uri
if has_unresolved_upserts:
+ if not has_unexplained_unresolved_upserts:
+ skip = SkippedMemoryOperation(
+ memory_type=(
+ file_content.memory_type
+ or file_content.extra_fields.get("memory_type")
+ or self.memory_type_from_uri(delete_uri)
+ or "unknown"
+ ),
+ uri=delete_uri,
+ reason_code=MemoryOperationSkipCode.DEPENDENT_DELETE_SUPPRESSED,
+ reason=(
+ "Delete was suppressed because the batch contains an "
+ "intentionally skipped upsert"
+ ),
+ )
+ result.add_skipped(skip)
+ tracer.info(
+ "Skipping dependent memory delete by resolution policy: "
+ f"memory_type={skip.memory_type} "
+ "reason_code=dependent_delete_suppressed"
+ )
+ continue
delete_error = ValueError(
"Skipped delete because batch contains unresolved upsert URIs"
)
diff --git a/openviking/session/memory/streaming_memory_updater.py b/openviking/session/memory/streaming_memory_updater.py
index 6c424c41cb..a8edbc3179 100644
--- a/openviking/session/memory/streaming_memory_updater.py
+++ b/openviking/session/memory/streaming_memory_updater.py
@@ -26,6 +26,7 @@
MemoryTypeSchema,
ResolvedOperation,
ResolvedOperations,
+ SkippedMemoryOperation,
StoredLink,
)
from openviking.session.memory.extract_loop import ExtractLoop
@@ -208,6 +209,7 @@ async def submit(self, request: MemoryUpdateRequest) -> StreamingMemoryUpdateRes
f"written_uris={scoped_result.apply_result.written_uris} "
f"edited_uris={scoped_result.apply_result.edited_uris} "
f"deleted_uris={scoped_result.apply_result.deleted_uris} "
+ f"skipped_reason_codes={_skipped_reason_codes(scoped_result.apply_result)} "
f"errors={scoped_result.apply_result.errors}",
console=self.config.trace_console,
)
@@ -398,6 +400,7 @@ async def _apply_append_only_request_now(
f"written_uris={apply_result.written_uris} "
f"edited_uris={apply_result.edited_uris} "
f"deleted_uris={apply_result.deleted_uris} "
+ f"skipped_reason_codes={_skipped_reason_codes(apply_result)} "
f"errors={apply_result.errors}",
console=self.config.trace_console,
)
@@ -449,6 +452,7 @@ async def _process_batch(
f"written_uris={apply_result.written_uris} "
f"edited_uris={apply_result.edited_uris} "
f"deleted_uris={apply_result.deleted_uris} "
+ f"skipped_reason_codes={_skipped_reason_codes(apply_result)} "
f"errors={apply_result.errors}",
console=self.config.trace_console,
)
@@ -612,12 +616,16 @@ def split_request_by_merge_group(
group_key = MemoryMergeGroupKey(peer_id=peer_id, memory_type=single_uri_op.memory_type)
upsert_groups.setdefault(group_key, []).append(single_uri_op)
- for file in list(operations.delete_file_contents or []):
- group_key = MemoryMergeGroupKey(
- peer_id=_peer_id_for_memory_file(file),
- memory_type=file.memory_type or "",
- )
- delete_groups.setdefault(group_key, []).append(file)
+ # Keep deletes in the same apply group as unresolved upserts so MemoryUpdater
+ # can observe both operations and suppress delete execution when an upsert is
+ # intentionally skipped.
+ if not passthrough_upserts:
+ for file in list(operations.delete_file_contents or []):
+ group_key = MemoryMergeGroupKey(
+ peer_id=_peer_id_for_memory_file(file),
+ memory_type=file.memory_type or "",
+ )
+ delete_groups.setdefault(group_key, []).append(file)
group_keys = list(dict.fromkeys(list(upsert_groups.keys()) + list(delete_groups.keys())))
grouped_requests: list[tuple[MemoryMergeGroupKey, MemoryUpdateRequest]] = []
@@ -650,6 +658,7 @@ def split_request_by_merge_group(
)
if passthrough_upserts:
+ passthrough_deletes = list(operations.delete_file_contents or [])
group_key = MemoryMergeGroupKey(peer_id=None, memory_type="")
grouped_requests.append(
(
@@ -658,10 +667,19 @@ def split_request_by_merge_group(
request,
operations=ResolvedOperations(
upsert_operations=passthrough_upserts,
- delete_file_contents=[],
+ delete_file_contents=passthrough_deletes,
errors=list(operations.errors or []),
resolved_links=[],
- delete_replacements={},
+ delete_replacements={
+ file.uri: replacement_uri
+ for file in passthrough_deletes
+ if file.uri
+ if (
+ replacement_uri := (
+ getattr(operations, "delete_replacements", {}) or {}
+ ).get(file.uri)
+ )
+ },
),
),
)
@@ -1610,6 +1628,7 @@ def scope_memory_update_result_to_submitter(
scoped_apply_result = _scope_apply_result_to_uris(
result.apply_result,
scoped_uris=scoped_uris,
+ scope=scope,
)
metadata = dict(result.metadata or {})
metadata.update(
@@ -1717,6 +1736,7 @@ def _scope_apply_result_to_uris(
apply_result: MemoryUpdateResult,
*,
scoped_uris: set[str],
+ scope: _MemorySubmitterScope,
) -> MemoryUpdateResult:
scoped = MemoryUpdateResult()
scoped.written_uris = [
@@ -1733,9 +1753,35 @@ def _scope_apply_result_to_uris(
for error in list(getattr(apply_result, "errors", []) or [])
if _apply_error_matches_scoped_uris(error, scoped_uris=scoped_uris)
]
+ scoped.skipped_operations = [
+ operation
+ for operation in list(getattr(apply_result, "skipped_operations", []) or [])
+ if _skipped_operation_matches_scope(
+ operation,
+ scope=scope,
+ scoped_uris=scoped_uris,
+ )
+ ]
return scoped
+def _skipped_operation_matches_scope(
+ operation: SkippedMemoryOperation,
+ *,
+ scope: _MemorySubmitterScope,
+ scoped_uris: set[str],
+) -> bool:
+ source = getattr(operation, "source", None)
+ if scope.extraction_id and getattr(source, "extraction_id", None) == scope.extraction_id:
+ return True
+ if scope.archive_uri and getattr(source, "archive_uri", None) == scope.archive_uri:
+ return True
+ if scope.session_id and getattr(source, "session_id", None) == scope.session_id:
+ return True
+ uri = str(getattr(operation, "uri", None) or "")
+ return bool(uri and uri in scoped_uris)
+
+
def _operation_matches_scope(op: ResolvedOperation, *, scope: _MemorySubmitterScope) -> bool:
if scope.extraction_id and scope.extraction_id in _operation_source_extraction_ids(op):
return True
@@ -1894,6 +1940,7 @@ def combine_streaming_memory_results(
combined_apply_result.written_uris.extend(result.apply_result.written_uris)
combined_apply_result.edited_uris.extend(result.apply_result.edited_uris)
combined_apply_result.deleted_uris.extend(result.apply_result.deleted_uris)
+ combined_apply_result.skipped_operations.extend(result.apply_result.skipped_operations)
combined_apply_result.errors.extend(result.apply_result.errors)
for key in ("batch_id", "batch_trace_id"):
if result.metadata.get(key):
@@ -1965,6 +2012,7 @@ def _make_isolation_handler(
allowed_memory_types=options.get("allowed_memory_types"),
allow_self=options.get("allow_self", True),
allowed_peer_ids=options.get("allowed_peer_ids"),
+ peer_memory_enabled=options.get("peer_memory_enabled"),
)
@@ -1972,6 +2020,13 @@ def _operation_count(operations: ResolvedOperations) -> int:
return len(operations.upsert_operations or []) + len(operations.delete_file_contents or [])
+def _skipped_reason_codes(result: MemoryUpdateResult) -> list[str]:
+ return [
+ operation.reason_code.value
+ for operation in list(getattr(result, "skipped_operations", []) or [])
+ ]
+
+
def _operation_lock_paths(
operations: ResolvedOperations,
viking_fs: Any | None,
diff --git a/openviking/session/session.py b/openviking/session/session.py
index ba1ca7bb59..f14a037c14 100644
--- a/openviking/session/session.py
+++ b/openviking/session/session.py
@@ -184,6 +184,7 @@ def _message_peer_ids(messages: List[Message]) -> set[str]:
@dataclass(frozen=True)
class _MemoryExtractionScope:
allow_self_memory: bool
+ peer_memory_enabled: bool
allowed_peer_ids: set[str]
include_session_skills: bool
memory_types: Optional[set[str]]
@@ -201,6 +202,7 @@ def _resolve_memory_extraction_scope(
return _MemoryExtractionScope(
allow_self_memory=allow_self_memory,
+ peer_memory_enabled=policy.peer_enabled,
allowed_peer_ids=allowed_peer_ids,
include_session_skills=config_session_skill_extraction_enabled and allow_self_memory,
memory_types=policy.memory_types,
@@ -2367,6 +2369,7 @@ async def _run_memory_extraction(
memories_extracted: Dict[str, int] = {}
usage_events_extracted = 0
extracted_skill_results: list[dict] = []
+ skipped_memory_operations: list[dict[str, Any]] = []
active_count_updated = 0
memory_diff_uri: Optional[str] = None
completed_memory_steps: Dict[str, set[str]] = {}
@@ -2551,6 +2554,7 @@ async def _run_recorded_memory_step(
),
)
self_memory_enabled = extraction_scope.allow_self_memory
+ peer_memory_enabled = extraction_scope.peer_memory_enabled
allowed_peer_ids = extraction_scope.allowed_peer_ids
long_term_memory_types = extraction_scope.memory_types
@@ -2598,6 +2602,7 @@ async def _run_long_term_memory_extraction() -> Any:
allowed_memory_types=long_term_memory_types,
agent_evolution_enabled=agent_evolution_enabled,
allow_self_memory=self_memory_enabled,
+ peer_memory_enabled=peer_memory_enabled,
allowed_peer_ids=allowed_peer_ids,
event_search_tags=event_search_tags,
)
@@ -2650,9 +2655,11 @@ async def _run_long_term_memory_extraction() -> Any:
if isinstance(result, dict):
target_contexts = list(result.get("contexts", []))
target_skills = list(result.get("session_skills", []))
+ target_skips = list(result.get("skipped_operations", []))
else:
target_contexts = list(result or [])
target_skills = []
+ target_skips = []
logger.info(
"Extracted %s memories for %s",
len(target_contexts),
@@ -2664,6 +2671,9 @@ async def _run_long_term_memory_extraction() -> Any:
memories_extracted[cat] = memories_extracted.get(cat, 0) + 1
if target_skills:
extracted_skill_results.extend(target_skills)
+ skipped_memory_operations.extend(
+ item for item in target_skips if isinstance(item, dict)
+ )
if total_extracted:
self._stats.memories_extracted += total_extracted
@@ -2754,6 +2764,10 @@ async def _run_long_term_memory_extraction() -> Any:
for item in extracted_skill_results
if isinstance(item, dict) and (item.get("uri") or item.get("root_uri"))
],
+ "memory_extraction": {
+ "skipped": len(skipped_memory_operations),
+ "skipped_operations": skipped_memory_operations,
+ },
"usage_events_extracted": usage_events_extracted,
"active_count_updated": active_count_updated,
"effective_memory_types": sorted(
diff --git a/tests/session/memory/test_memory_diff.py b/tests/session/memory/test_memory_diff.py
index a55bb0abd7..6d6a2a8442 100644
--- a/tests/session/memory/test_memory_diff.py
+++ b/tests/session/memory/test_memory_diff.py
@@ -477,7 +477,14 @@ class TestMemoryDiffStructure:
def test_memory_diff_structure(self):
"""Verify memory_diff.json structure."""
# This test validates the expected structure
- expected_keys = ["archive_uri", "trace_id", "extracted_at", "operations", "summary"]
+ expected_keys = [
+ "archive_uri",
+ "trace_id",
+ "extracted_at",
+ "operations",
+ "skipped_operations",
+ "summary",
+ ]
# We verify this through the actual implementation tests above
# This is a placeholder for documentation
@@ -487,6 +494,7 @@ def test_memory_diff_structure(self):
"trace_id",
"extracted_at",
"operations",
+ "skipped_operations",
"summary",
}
)
diff --git a/tests/session/memory/test_memory_isolation_handler.py b/tests/session/memory/test_memory_isolation_handler.py
index 1ec666af8c..341f1538e7 100644
--- a/tests/session/memory/test_memory_isolation_handler.py
+++ b/tests/session/memory/test_memory_isolation_handler.py
@@ -11,6 +11,7 @@
from openviking.message.message import Message
from openviking.message.part import TextPart
from openviking.server.identity import RequestContext, Role
+from openviking.session.memory.dataclass import MemoryOperationSkipCode
from openviking.session.memory.memory_isolation_handler import (
MemoryIsolationHandler,
)
@@ -876,9 +877,76 @@ def test_calculate_memory_uris_unallowed_peer_id_does_not_fallback(self, mock_ge
uris = handler.calculate_memory_uris(schema, operation, extract_ctx)
assert uris == []
+ assert operation.resolution_skip is not None
+ assert operation.resolution_skip.reason_code == MemoryOperationSkipCode.PEER_NOT_ALLOWED
assert "peer_id" not in operation.memory_fields
mock_generate_uri.assert_not_called()
+ def test_calculate_memory_uris_classifies_peer_policy_skip(self):
+ from openviking.session.memory.dataclass import MemoryTypeSchema, ResolvedOperation
+
+ ctx = create_ctx(user_id="support_bot")
+ extract_ctx = create_mock_extract_context(
+ [create_message("user", peer_id="web-visitor-alice")]
+ )
+ handler = MemoryIsolationHandler(
+ ctx,
+ extract_ctx,
+ allow_self=False,
+ allowed_peer_ids=set(),
+ peer_memory_enabled=False,
+ )
+ operation = ResolvedOperation(
+ old_memory_file_content=None,
+ memory_fields={"peer_id": "web-visitor-alice"},
+ memory_type="preferences",
+ uris=[],
+ )
+
+ uris = handler.calculate_memory_uris(
+ MemoryTypeSchema(
+ memory_type="preferences",
+ filename_template="preferences.md",
+ directory="viking://user/{user_space}/memories",
+ ),
+ operation,
+ extract_ctx,
+ )
+
+ assert uris == []
+ assert operation.resolution_skip is not None
+ assert operation.resolution_skip.reason_code == (
+ MemoryOperationSkipCode.PEER_MEMORY_DISABLED
+ )
+
+ def test_calculate_memory_uris_classifies_invalid_ranges(self):
+ from openviking.session.memory.dataclass import MemoryTypeSchema, ResolvedOperation
+
+ ctx = create_ctx(user_id="support_bot")
+ extract_ctx = create_mock_extract_context([create_message("user")])
+ handler = MemoryIsolationHandler(ctx, extract_ctx)
+ operation = ResolvedOperation(
+ old_memory_file_content=None,
+ memory_fields={"ranges": "10-20"},
+ memory_type="preferences",
+ uris=[],
+ )
+
+ uris = handler.calculate_memory_uris(
+ MemoryTypeSchema(
+ memory_type="preferences",
+ filename_template="preferences.md",
+ directory="viking://user/{user_space}/memories",
+ ),
+ operation,
+ extract_ctx,
+ )
+
+ assert uris == []
+ assert operation.resolution_skip is not None
+ assert operation.resolution_skip.reason_code == MemoryOperationSkipCode.INVALID_RANGES
+ extract_ctx.read_message_ranges.assert_not_called()
+
@patch("openviking.session.memory.memory_isolation_handler.generate_uri")
def test_calculate_memory_uris_missing_peer_id_prefers_self_when_allowed(
self, mock_generate_uri
@@ -1196,5 +1264,7 @@ def test_calculate_memory_uris_invalid_peer_id_does_not_fallback(self, mock_gene
uris = handler.calculate_memory_uris(schema, operation, extract_ctx)
assert uris == []
+ assert operation.resolution_skip is not None
+ assert operation.resolution_skip.reason_code == MemoryOperationSkipCode.INVALID_PEER_ID
assert "peer_id" not in operation.memory_fields
mock_generate_uri.assert_not_called()
diff --git a/tests/session/memory/test_memory_updater.py b/tests/session/memory/test_memory_updater.py
index 6d623011d6..581446ff82 100644
--- a/tests/session/memory/test_memory_updater.py
+++ b/tests/session/memory/test_memory_updater.py
@@ -15,6 +15,8 @@
from openviking.session.memory.dataclass import (
MemoryField,
MemoryFile,
+ MemoryOperationSkip,
+ MemoryOperationSkipCode,
MemoryOperationSource,
MemoryTypeSchema,
ResolvedOperation,
@@ -480,6 +482,49 @@ async def test_apply_operations_isolates_unresolved_uris(self):
"Skipping unresolved memory operation: events(page_id=102): Missing resolved URI"
)
+ @pytest.mark.asyncio
+ async def test_apply_operations_reports_expected_empty_uri_as_skip(self):
+ updater = MemoryUpdater(registry=MagicMock())
+ updater._get_viking_fs = MagicMock(return_value=MagicMock())
+ updater._apply_upsert = AsyncMock(return_value=None)
+ updater._sync_resource_refs_for_result = AsyncMock()
+ updater._vectorize_memories = AsyncMock()
+ updater.generate_overview = AsyncMock()
+ operation = ResolvedOperation(
+ memory_fields={"peer_id": "web-visitor-alice"},
+ memory_type="preferences",
+ uris=[],
+ page_id=102,
+ resolution_skip=MemoryOperationSkip(
+ reason_code=MemoryOperationSkipCode.PEER_NOT_ALLOWED,
+ reason="Target peer is outside the allowed memory scope",
+ ),
+ )
+ operations = ResolvedOperations(
+ upsert_operations=[operation],
+ delete_file_contents=[],
+ errors=[],
+ )
+ ctx = RequestContext(user=UserIdentifier("acme", "alice"), role=Role.USER)
+
+ with (
+ patch("openviking.session.memory.memory_updater.tracer.info") as tracer_info,
+ patch("openviking.session.memory.memory_updater.tracer.error") as tracer_error,
+ ):
+ result = await updater.apply_operations(operations=operations, ctx=ctx)
+
+ assert result.errors == []
+ assert len(result.skipped_operations) == 1
+ assert result.skipped_operations[0].reason_code == (
+ MemoryOperationSkipCode.PEER_NOT_ALLOWED
+ )
+ updater._apply_upsert.assert_not_awaited()
+ tracer_error.assert_not_called()
+ tracer_info.assert_any_call(
+ "Skipping memory operation by resolution policy: "
+ "memory_type=preferences page_id=102 reason_code=peer_not_allowed"
+ )
+
@pytest.mark.asyncio
async def test_apply_operations_skips_deletes_when_replacement_uri_is_unresolved(self):
registry = MagicMock()
@@ -534,6 +579,46 @@ async def test_apply_operations_skips_deletes_when_replacement_uri_is_unresolved
]
updater._apply_delete.assert_not_awaited()
+ @pytest.mark.asyncio
+ async def test_apply_operations_safely_suppresses_delete_for_expected_uri_skip(self):
+ updater = MemoryUpdater(registry=MagicMock())
+ updater._get_viking_fs = MagicMock(return_value=MagicMock())
+ updater._apply_upsert = AsyncMock(return_value=None)
+ updater._apply_delete = AsyncMock()
+ updater._sync_resource_refs_for_result = AsyncMock()
+ updater._vectorize_memories = AsyncMock()
+ updater.generate_overview = AsyncMock()
+ old_uri = "viking://user/alice/memories/preferences/old.md"
+ operations = ResolvedOperations(
+ upsert_operations=[
+ ResolvedOperation(
+ memory_fields={"ranges": "99"},
+ memory_type="preferences",
+ uris=[],
+ page_id=102,
+ resolution_skip=MemoryOperationSkip(
+ reason_code=MemoryOperationSkipCode.INVALID_RANGES,
+ reason="Message ranges are malformed or out of bounds",
+ ),
+ )
+ ],
+ delete_file_contents=[
+ MemoryFile(uri=old_uri, extra_fields={"memory_type": "preferences"})
+ ],
+ errors=[],
+ )
+ ctx = RequestContext(user=UserIdentifier("acme", "alice"), role=Role.USER)
+
+ result = await updater.apply_operations(operations=operations, ctx=ctx)
+
+ assert result.errors == []
+ assert [item.reason_code for item in result.skipped_operations] == [
+ MemoryOperationSkipCode.INVALID_RANGES,
+ MemoryOperationSkipCode.DEPENDENT_DELETE_SUPPRESSED,
+ ]
+ assert result.skipped_operations[1].memory_type == "preferences"
+ updater._apply_delete.assert_not_awaited()
+
@pytest.mark.asyncio
async def test_apply_operations_matches_overview_directory_from_resolved_user_uri(self):
"""Overview generation should use the resolved user memory directory."""
diff --git a/tests/session/memory/test_streaming_memory_updater.py b/tests/session/memory/test_streaming_memory_updater.py
index 3b2772a0c0..64e114f0bb 100644
--- a/tests/session/memory/test_streaming_memory_updater.py
+++ b/tests/session/memory/test_streaming_memory_updater.py
@@ -13,10 +13,12 @@
from openviking.session.memory.dataclass import (
MemoryField,
MemoryFile,
+ MemoryOperationSkipCode,
MemoryOperationSource,
MemoryTypeSchema,
ResolvedOperation,
ResolvedOperations,
+ SkippedMemoryOperation,
StoredLink,
)
from openviking.session.memory.memory_type_registry import MemoryTypeRegistry
@@ -715,8 +717,7 @@ async def fake_merge_memory_operations(**kwargs):
else:
kind = "update"
assert all(
- op.old_memory_file_content is not None
- for op in operations.upsert_operations
+ op.old_memory_file_content is not None for op in operations.upsert_operations
)
assert kwargs["force_merge"] is True
entered.add(kind)
@@ -795,6 +796,22 @@ def test_scope_memory_update_result_to_submitter_filters_shared_batch_by_source(
apply_result = MemoryUpdateResult()
apply_result.add_written(op_a.uris[0])
apply_result.add_written(op_b.uris[0])
+ apply_result.add_skipped(
+ SkippedMemoryOperation(
+ memory_type="preferences",
+ reason_code=MemoryOperationSkipCode.PEER_NOT_ALLOWED,
+ reason="Target peer is outside the allowed memory scope",
+ source=MemoryOperationSource(extraction_id="extract_a"),
+ )
+ )
+ apply_result.add_skipped(
+ SkippedMemoryOperation(
+ memory_type="preferences",
+ reason_code=MemoryOperationSkipCode.PEER_MEMORY_DISABLED,
+ reason="Peer memory writes are disabled",
+ source=MemoryOperationSource(extraction_id="extract_b"),
+ )
+ )
batch_result = StreamingMemoryUpdateResult(
operations=ResolvedOperations(
upsert_operations=[op_a, op_b],
@@ -822,6 +839,10 @@ def test_scope_memory_update_result_to_submitter_filters_shared_batch_by_source(
assert scoped.metadata["batch_request_count"] == 2
assert scoped.metadata["scoped_to_source_extraction_id"] == "extract_a"
assert scoped.apply_result.written_uris == [op_a.uris[0]]
+ assert len(scoped.apply_result.skipped_operations) == 1
+ assert scoped.apply_result.skipped_operations[0].reason_code == (
+ MemoryOperationSkipCode.PEER_NOT_ALLOWED
+ )
assert scoped.operations.upsert_operations == [op_a]
assert scoped.metadata["unscoped_written_uris"] == [op_a.uris[0], op_b.uris[0]]
@@ -866,6 +887,29 @@ def test_split_request_by_merge_group_groups_by_peer_and_memory_type():
]
+def test_split_request_keeps_deletes_with_unresolved_upserts():
+ unresolved = _note_op("replacement")
+ unresolved.uris = []
+ old_file = _note_delete_file("old")
+ request = MemoryUpdateRequest(
+ operations=ResolvedOperations(
+ upsert_operations=[unresolved],
+ delete_file_contents=[old_file],
+ errors=[],
+ ),
+ messages=[],
+ ctx=_ctx(),
+ )
+
+ grouped = split_request_by_merge_group(request)
+
+ assert len(grouped) == 1
+ group_key, group_request = grouped[0]
+ assert group_key == MemoryMergeGroupKey(peer_id=None, memory_type="")
+ assert group_request.operations.upsert_operations == [unresolved]
+ assert group_request.operations.delete_file_contents == [old_file]
+
+
def test_split_request_by_merge_group_infers_peer_from_uri_when_field_missing():
peer_uri = "viking://user/u/peers/conv-42/memories/notes/peer_note.md"
op = ResolvedOperation(
diff --git a/tests/session/test_commit_skill_inference.py b/tests/session/test_commit_skill_inference.py
index 37b8aaa5fc..35cd81ef74 100644
--- a/tests/session/test_commit_skill_inference.py
+++ b/tests/session/test_commit_skill_inference.py
@@ -77,6 +77,27 @@ def test_v3_extraction_response_returns_session_skills():
}
+def test_v3_extraction_response_returns_skipped_operations():
+ skipped = {
+ "memory_type": "preferences",
+ "reason_code": "invalid_ranges",
+ "reason": "Message ranges are malformed or out of bounds",
+ }
+
+ result = _v3_extraction_response(
+ contexts=[],
+ train_result={},
+ archive_uri="viking://sessions/s1/history/archive_001",
+ skipped_operations=[skipped],
+ )
+
+ assert result == {
+ "contexts": [],
+ "session_skills": [],
+ "skipped_operations": [skipped],
+ }
+
+
def test_session_skill_operations_dedup_duplicate_creates():
result = dedup_session_skill_operations(_build_duplicate_session_skill_operations())
diff --git a/tests/session/test_compressor_v3.py b/tests/session/test_compressor_v3.py
index be59bdb4c0..494360d2bc 100644
--- a/tests/session/test_compressor_v3.py
+++ b/tests/session/test_compressor_v3.py
@@ -3,6 +3,7 @@
from __future__ import annotations
+import inspect
from types import SimpleNamespace
from unittest.mock import AsyncMock
@@ -21,8 +22,10 @@
)
from openviking.session.memory.dataclass import (
MemoryFile,
+ MemoryOperationSkipCode,
ResolvedOperation,
ResolvedOperations,
+ SkippedMemoryOperation,
StoredLink,
)
from openviking.session.memory.memory_updater import MemoryUpdateResult
@@ -95,6 +98,64 @@ def test_factory_ignores_deprecated_memory_version():
)
+def test_extract_long_term_memories_preserves_legacy_positional_parameter_order():
+ parameter_names = list(
+ inspect.signature(SessionCompressorV3.extract_long_term_memories).parameters
+ )
+
+ assert parameter_names[-4:] == [
+ "allow_self_memory",
+ "allowed_peer_ids",
+ "event_search_tags",
+ "peer_memory_enabled",
+ ]
+
+
+@pytest.mark.asyncio
+async def test_memory_diff_includes_intentionally_skipped_operations(monkeypatch):
+ monkeypatch.setattr(
+ "openviking.session.compressor_v3.get_viking_fs",
+ lambda: SimpleNamespace(),
+ )
+ compressor = SessionCompressorV3(vikingdb=None)
+ result = MemoryUpdateResult()
+ result.add_skipped(
+ SkippedMemoryOperation(
+ memory_type="events",
+ page_id=101,
+ reason_code=MemoryOperationSkipCode.INVALID_RANGES,
+ reason="No valid event range could be resolved",
+ )
+ )
+
+ diff = await compressor._build_memory_diff(
+ result=result,
+ operations=ResolvedOperations(
+ upsert_operations=[],
+ delete_file_contents=[],
+ errors=[],
+ ),
+ viking_fs=SimpleNamespace(),
+ ctx=_ctx(),
+ archive_uri="viking://user/u/sessions/s1/history/archive_001",
+ )
+
+ assert diff["skipped_operations"] == [
+ {
+ "memory_type": "events",
+ "page_id": 101,
+ "reason_code": "invalid_ranges",
+ "reason": "No valid event range could be resolved",
+ }
+ ]
+ assert diff["summary"] == {
+ "total_adds": 0,
+ "total_updates": 0,
+ "total_deletes": 0,
+ "total_skipped": 1,
+ }
+
+
@pytest.mark.asyncio
async def test_v3_skips_agent_training_when_agent_evolution_is_disabled(monkeypatch):
monkeypatch.setattr(
@@ -108,12 +169,19 @@ async def test_v3_skips_agent_training_when_agent_evolution_is_disabled(monkeypa
cases=[_training_case()],
memory_diff={"operations": {}},
case_uri_by_name={},
+ skipped_operations=[
+ {
+ "memory_type": "profile",
+ "reason_code": "peer_memory_disabled",
+ "reason": "Peer memory writes are disabled",
+ }
+ ],
)
)
compressor.train_from_extracted_cases = AsyncMock()
compressor._write_final_memory_diff = AsyncMock()
- await compressor.extract_long_term_memories(
+ result = await compressor.extract_long_term_memories(
messages=_messages(),
ctx=_ctx(),
allowed_memory_types={"cases", "profile"},
@@ -121,6 +189,13 @@ async def test_v3_skips_agent_training_when_agent_evolution_is_disabled(monkeypa
)
compressor.train_from_extracted_cases.assert_not_awaited()
+ assert result["skipped_operations"] == [
+ {
+ "memory_type": "profile",
+ "reason_code": "peer_memory_disabled",
+ "reason": "Peer memory writes are disabled",
+ }
+ ]
@pytest.mark.asyncio
@@ -1065,7 +1140,20 @@ async def fake_write_training_case_memory(**kwargs):
"updates": [],
"deletes": [],
},
- "summary": {"total_adds": 1, "total_updates": 0, "total_deletes": 0},
+ "skipped_operations": [
+ {
+ "memory_type": "preferences",
+ "page_id": 102,
+ "reason_code": "peer_not_allowed",
+ "reason": "Target peer is outside the allowed memory scope",
+ }
+ ],
+ "summary": {
+ "total_adds": 1,
+ "total_updates": 0,
+ "total_deletes": 0,
+ "total_skipped": 1,
+ },
},
)
@@ -1118,7 +1206,20 @@ async def fake_train_from_extracted_cases(**kwargs):
"trajectories",
]
assert [item["memory_type"] for item in diff["operations"]["updates"]] == ["experiences"]
- assert diff["summary"] == {"total_adds": 2, "total_updates": 1, "total_deletes": 0}
+ assert diff["skipped_operations"] == [
+ {
+ "memory_type": "preferences",
+ "page_id": 102,
+ "reason_code": "peer_not_allowed",
+ "reason": "Target peer is outside the allowed memory scope",
+ }
+ ]
+ assert diff["summary"] == {
+ "total_adds": 2,
+ "total_updates": 1,
+ "total_deletes": 0,
+ "total_skipped": 1,
+ }
@pytest.mark.asyncio
@@ -1190,7 +1291,12 @@ async def read_file(self, uri, ctx=None):
archive_uri=archive_uri,
)
- assert diff["summary"] == {"total_adds": 1, "total_updates": 1, "total_deletes": 0}
+ assert diff["summary"] == {
+ "total_adds": 1,
+ "total_updates": 1,
+ "total_deletes": 0,
+ "total_skipped": 0,
+ }
assert diff["operations"]["adds"][0]["memory_type"] == "trajectories"
update = diff["operations"]["updates"][0]
assert update["memory_type"] == "experiences"
@@ -1287,7 +1393,12 @@ async def read_file(self, uri, ctx=None):
archive_uri=archive_uri,
)
- assert diff["summary"] == {"total_adds": 2, "total_updates": 0, "total_deletes": 0}
+ assert diff["summary"] == {
+ "total_adds": 2,
+ "total_updates": 0,
+ "total_deletes": 0,
+ "total_skipped": 0,
+ }
assert [op["uri"] for op in diff["operations"]["adds"]] == [traj_a, exp_a]
diff --git a/tests/session/test_session_commit.py b/tests/session/test_session_commit.py
index 969b8ce555..2530a36302 100644
--- a/tests/session/test_session_commit.py
+++ b/tests/session/test_session_commit.py
@@ -115,6 +115,42 @@ async def extract_long_term_memories(**kwargs):
# Wait for semantic/embedding queues
await service.resources.wait_processed(timeout=60.0)
+ async def test_commit_task_reports_intentionally_skipped_memory_operations(
+ self,
+ session_with_messages: Session,
+ ):
+ session_with_messages._session_compressor.extract_long_term_memories = AsyncMock(
+ return_value={
+ "contexts": [],
+ "session_skills": [],
+ "skipped_operations": [
+ {
+ "memory_type": "preferences",
+ "page_id": 102,
+ "reason_code": "peer_not_allowed",
+ "reason": "Target peer is outside the allowed memory scope",
+ }
+ ],
+ }
+ )
+
+ commit_result = await session_with_messages.commit_async()
+ task_result = await _wait_for_task(commit_result["task_id"])
+
+ assert commit_result["status"] == "accepted"
+ assert task_result["status"] == "completed"
+ assert task_result["result"]["memory_extraction"] == {
+ "skipped": 1,
+ "skipped_operations": [
+ {
+ "memory_type": "preferences",
+ "page_id": 102,
+ "reason_code": "peer_not_allowed",
+ "reason": "Target peer is outside the allowed memory scope",
+ }
+ ],
+ }
+
async def test_commit_default_disables_agent_memory_but_keeps_archive(
self, session_with_messages: Session
):
@@ -340,6 +376,7 @@ async def fake_extract(
ctx,
allowed_memory_types,
allow_self_memory=True,
+ peer_memory_enabled=True,
allowed_peer_ids=None,
**kwargs,
):
@@ -348,6 +385,7 @@ async def fake_extract(
{
"allowed_memory_types": set(allowed_memory_types or set()),
"allow_self_memory": allow_self_memory,
+ "peer_memory_enabled": peer_memory_enabled,
"allowed_peer_ids": set(allowed_peer_ids or set()),
"roles": [message.role for message in messages],
"peer_ids": [message.peer_id for message in messages],
@@ -386,6 +424,7 @@ async def fake_extract(
"profile",
},
"allow_self_memory": False,
+ "peer_memory_enabled": True,
"allowed_peer_ids": {"web-visitor-alice"},
"roles": ["user", "assistant"],
"peer_ids": ["web-visitor-alice", "web-visitor-alice"],