mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-27 17:41:56 +00:00
922 lines
32 KiB
Python
922 lines
32 KiB
Python
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import unicodedata
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from functools import lru_cache
|
|
from typing import Any, Literal
|
|
|
|
from app.permission_policy import redact_action_arguments
|
|
from app.run_journal import (
|
|
InvalidRunTransitionError,
|
|
MemoryEntryRecord,
|
|
MemoryMutationResult,
|
|
MemoryScopeStateRecord,
|
|
SQLiteRunJournal,
|
|
get_default_run_journal,
|
|
)
|
|
from app.run_journal.memory_policy import assert_memory_entry_policy
|
|
|
|
try:
|
|
import tiktoken
|
|
except ImportError: # pragma: no cover - production bundle includes it
|
|
tiktoken = None
|
|
|
|
|
|
_TOKEN_LIMITS = {"project": 1024, "space": 640, "user": 384}
|
|
_QUERY_TOKEN_RE = re.compile(r"[\w.-]+", re.UNICODE)
|
|
_HISTORY_ARTIFACT_FIELDS = frozenset(
|
|
{
|
|
"artifact_id",
|
|
"filename",
|
|
"relativePath",
|
|
"changeType",
|
|
"size",
|
|
"mimeType",
|
|
"scan_status",
|
|
"truncated",
|
|
"artifact_count",
|
|
"manifest_digest",
|
|
}
|
|
)
|
|
|
|
|
|
def _history_payload_projection(
|
|
event_type: str, payload: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""Return the model-visible, normalized History projection.
|
|
|
|
Canonical Run events remain untouched in SQLite. The Agent History Tool
|
|
gets only redacted arguments, connector/tool outcome metadata, and
|
|
Artifact metadata. Raw Connector bodies, tool results and local Artifact
|
|
paths are deliberately unavailable through this principal.
|
|
"""
|
|
|
|
redacted = redact_action_arguments(payload)
|
|
if event_type.startswith("tool."):
|
|
return {
|
|
key: value
|
|
for key, value in redacted.items()
|
|
if key
|
|
in {
|
|
"tool_call_id",
|
|
"attempt_id",
|
|
"tool_name",
|
|
"safety_class",
|
|
"status",
|
|
"outcome",
|
|
"timeout_reason",
|
|
"request",
|
|
}
|
|
}
|
|
if event_type.startswith("artifact."):
|
|
if event_type == "artifact.manifest.finalized":
|
|
artifacts = redacted.get("artifacts")
|
|
projected_artifacts = (
|
|
[
|
|
{
|
|
key: value
|
|
for key, value in artifact.items()
|
|
if key in _HISTORY_ARTIFACT_FIELDS
|
|
}
|
|
for artifact in artifacts
|
|
if isinstance(artifact, dict)
|
|
]
|
|
if isinstance(artifacts, list)
|
|
else []
|
|
)
|
|
return {
|
|
key: value
|
|
for key, value in redacted.items()
|
|
if key in _HISTORY_ARTIFACT_FIELDS
|
|
} | {"artifacts": projected_artifacts}
|
|
return {
|
|
key: value
|
|
for key, value in redacted.items()
|
|
if key in _HISTORY_ARTIFACT_FIELDS
|
|
}
|
|
return redacted
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HistoryQueryResult:
|
|
citation_id: str
|
|
journal_cursor: int
|
|
event_id: str
|
|
run_id: str
|
|
event_type: str
|
|
content: dict[str, Any]
|
|
source_trust: str
|
|
created_at: float
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HistoryQueryPage:
|
|
source: str
|
|
project_id: str
|
|
complete: bool
|
|
freshness: str
|
|
watermark_kind: str
|
|
source_watermark: str
|
|
available_through_watermark: str
|
|
items: tuple[HistoryQueryResult, ...]
|
|
next_cursor: str
|
|
redactions: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MemoryConsolidationResult:
|
|
scope_state: MemoryScopeStateRecord
|
|
removed_memory_ids: tuple[str, ...]
|
|
retained_memory_ids: tuple[str, ...]
|
|
tokens_released: int
|
|
|
|
|
|
class LightweightMemoryService:
|
|
"""Typed boundary shared by local UI and Agent tools.
|
|
|
|
The service never reads legacy LocalMemory as History. Canonical events
|
|
come only from the RunJournal Project cursor.
|
|
"""
|
|
|
|
def __init__(self, journal: SQLiteRunJournal) -> None:
|
|
self._journal = journal
|
|
|
|
@property
|
|
def journal(self) -> SQLiteRunJournal:
|
|
return self._journal
|
|
|
|
def scope(self, scope_type: str, scope_id: str) -> MemoryScopeStateRecord:
|
|
return self._journal.ensure_memory_scope_state(
|
|
scope_type,
|
|
scope_id,
|
|
token_limit=_TOKEN_LIMITS[scope_type],
|
|
)
|
|
|
|
def list_entries(
|
|
self,
|
|
scope_type: str,
|
|
scope_id: str,
|
|
*,
|
|
include_deleted: bool = False,
|
|
) -> tuple[MemoryEntryRecord, ...]:
|
|
self.scope(scope_type, scope_id)
|
|
return tuple(
|
|
self._journal.list_memory_entries(
|
|
scope_type,
|
|
scope_id,
|
|
include_deleted=include_deleted,
|
|
)
|
|
)
|
|
|
|
def resolve_reconciliation(
|
|
self,
|
|
reconciliation_id: str,
|
|
*,
|
|
account_owner_id: str,
|
|
decision: Literal["local", "cloud", "dismiss"],
|
|
):
|
|
"""Resolve a visible cross-device conflict without silent LWW."""
|
|
|
|
item = self._journal.get_memory_reconciliation_item(reconciliation_id)
|
|
if item is None:
|
|
raise KeyError(reconciliation_id)
|
|
if item.account_owner_id != account_owner_id:
|
|
raise PermissionError(
|
|
"Memory reconciliation belongs to another account"
|
|
)
|
|
if decision in {"local", "dismiss"}:
|
|
resolution = (
|
|
"accepted_local" if decision == "local" else "dismissed"
|
|
)
|
|
return self._journal.resolve_memory_reconciliation_item(
|
|
reconciliation_id,
|
|
resolution=resolution,
|
|
)
|
|
cloud = item.cloud_entry
|
|
existing = self._require_entry(item.memory_id)
|
|
if (
|
|
existing.scope_type != item.scope_type
|
|
or existing.scope_id != item.scope_id
|
|
):
|
|
raise PermissionError("Memory reconciliation scope changed")
|
|
current_shape = {
|
|
"kind": existing.kind,
|
|
"content": existing.content,
|
|
"priority": existing.priority,
|
|
"token_count": existing.token_count,
|
|
"created_by": existing.created_by,
|
|
"source_trust": existing.source_trust,
|
|
"sensitivity": existing.sensitivity,
|
|
"source_refs": list(existing.source_refs),
|
|
"deleted_at": existing.deleted_at,
|
|
}
|
|
if current_shape != item.local_entry:
|
|
raise InvalidRunTransitionError(
|
|
"Local Memory changed after this review item was created; "
|
|
"refresh conflicts before choosing the Cloud version"
|
|
)
|
|
cloud_deleted = cloud.get("deleted_at") is not None
|
|
request_prefix = f"memory-reconciliation:{reconciliation_id}"
|
|
if cloud_deleted:
|
|
if existing.deleted_at is None:
|
|
self._journal.apply_memory_mutation(
|
|
mutation_id=stable_memory_id(
|
|
"mutation", request_prefix, "remove"
|
|
),
|
|
idempotency_key=f"{request_prefix}:remove",
|
|
operation="remove",
|
|
scope_type=existing.scope_type,
|
|
scope_id=existing.scope_id,
|
|
memory_id=existing.memory_id,
|
|
expected_version=existing.version,
|
|
actor_type="user",
|
|
reason="Accepted Cloud Memory deletion in Memory Center",
|
|
)
|
|
else:
|
|
content = str(cloud.get("content") or "")
|
|
kind = str(cloud.get("kind") or "")
|
|
source_trust = str(cloud.get("source_trust") or "")
|
|
cloud_created_by = str(cloud.get("created_by") or "")
|
|
cloud_refs = tuple(cloud.get("source_refs") or ())
|
|
assert_memory_entry_policy(
|
|
kind=kind,
|
|
content=content,
|
|
created_by=cloud_created_by,
|
|
source_trust=source_trust,
|
|
confirmed_by_user=bool(cloud.get("confirmed_by_user")),
|
|
source_refs=cloud_refs,
|
|
cloud_projection=True,
|
|
)
|
|
if existing.deleted_at is not None:
|
|
self._journal.apply_memory_mutation(
|
|
mutation_id=stable_memory_id(
|
|
"mutation", request_prefix, "restore"
|
|
),
|
|
idempotency_key=f"{request_prefix}:restore",
|
|
operation="restore",
|
|
scope_type=existing.scope_type,
|
|
scope_id=existing.scope_id,
|
|
memory_id=existing.memory_id,
|
|
expected_version=existing.version,
|
|
actor_type="user",
|
|
reason="Accepted active Cloud Memory in Memory Center",
|
|
)
|
|
existing = self._require_entry(item.memory_id)
|
|
local_semantic = (
|
|
existing.kind,
|
|
existing.content,
|
|
existing.priority,
|
|
existing.created_by,
|
|
existing.source_trust,
|
|
existing.sensitivity,
|
|
existing.source_refs,
|
|
)
|
|
cloud_semantic = (
|
|
kind,
|
|
content,
|
|
str(cloud.get("priority") or "normal"),
|
|
cloud_created_by,
|
|
source_trust,
|
|
str(cloud.get("sensitivity") or "normal"),
|
|
cloud_refs,
|
|
)
|
|
if local_semantic != cloud_semantic:
|
|
self._journal.apply_memory_mutation(
|
|
mutation_id=stable_memory_id(
|
|
"mutation", request_prefix, "replace"
|
|
),
|
|
idempotency_key=f"{request_prefix}:replace",
|
|
operation="replace",
|
|
scope_type=existing.scope_type,
|
|
scope_id=existing.scope_id,
|
|
memory_id=existing.memory_id,
|
|
expected_version=existing.version,
|
|
actor_type="system",
|
|
reason="Accepted Cloud Memory content in Memory Center",
|
|
content=content,
|
|
kind=kind,
|
|
priority=str(cloud.get("priority") or "normal"),
|
|
token_count=count_tokens(content),
|
|
created_by=cloud_created_by,
|
|
source_trust=source_trust,
|
|
sensitivity=str(cloud.get("sensitivity") or "normal"),
|
|
source_refs=cloud_refs,
|
|
)
|
|
return self._journal.resolve_memory_reconciliation_item(
|
|
reconciliation_id,
|
|
resolution="accepted_cloud",
|
|
)
|
|
|
|
def create_entry(
|
|
self,
|
|
*,
|
|
scope_type: str,
|
|
scope_id: str,
|
|
kind: str,
|
|
content: str,
|
|
actor_type: Literal["agent", "user", "extractor", "importer"],
|
|
reason: str,
|
|
source_trust: str,
|
|
source_refs: tuple[str, ...] = (),
|
|
priority: str = "normal",
|
|
sensitivity: str = "normal",
|
|
memory_id: str | None = None,
|
|
request_id: str | None = None,
|
|
actor_id: str | None = None,
|
|
run_id: str | None = None,
|
|
activity_id: str | None = None,
|
|
decision_id: str | None = None,
|
|
confirmed_by_user_action: bool = False,
|
|
adopted_by_user: bool = False,
|
|
reviewed_source_memory_id: str | None = None,
|
|
) -> MemoryMutationResult:
|
|
source_refs = tuple(dict.fromkeys(source_refs))
|
|
if adopted_by_user and not (
|
|
actor_type == "agent"
|
|
and confirmed_by_user_action
|
|
and decision_id
|
|
and reviewed_source_memory_id
|
|
):
|
|
raise PermissionError(
|
|
"User-adopted Memory requires an exact durable review"
|
|
)
|
|
entry_created_by = "user" if adopted_by_user else actor_type
|
|
entry_source_trust = (
|
|
"user_confirmed" if adopted_by_user else source_trust
|
|
)
|
|
self._assert_mutation_policy(
|
|
scope_type=scope_type,
|
|
kind=kind,
|
|
content=content,
|
|
actor_type=actor_type,
|
|
source_trust=entry_source_trust,
|
|
confirmed_by_user_action=confirmed_by_user_action,
|
|
entry_created_by=entry_created_by,
|
|
)
|
|
if actor_type == "agent" and not adopted_by_user:
|
|
self._assert_agent_provenance(
|
|
scope_type=scope_type,
|
|
scope_id=scope_id,
|
|
source_trust=entry_source_trust,
|
|
source_refs=source_refs,
|
|
)
|
|
state = self.scope(scope_type, scope_id)
|
|
if (
|
|
actor_type in {"agent", "extractor"}
|
|
and state.current_token_count >= state.token_limit * 0.9
|
|
):
|
|
raise InvalidRunTransitionError(
|
|
"Memory is at least 90% full. Search the current Memory and "
|
|
"replace, remove, or consolidate an existing item before "
|
|
"adding another one. Canonical History remains searchable."
|
|
)
|
|
resolved_request_id = request_id or f"memreq_{uuid.uuid4().hex}"
|
|
resolved_memory_id = memory_id or stable_memory_id(
|
|
"entry", scope_type, scope_id, resolved_request_id
|
|
)
|
|
return self._journal.apply_memory_mutation(
|
|
mutation_id=stable_memory_id("mutation", resolved_request_id),
|
|
idempotency_key=resolved_request_id,
|
|
operation="add",
|
|
scope_type=scope_type,
|
|
scope_id=scope_id,
|
|
memory_id=resolved_memory_id,
|
|
actor_type=actor_type,
|
|
actor_id=actor_id,
|
|
reason=reason,
|
|
content=content,
|
|
kind=kind,
|
|
priority=priority,
|
|
token_count=count_tokens(content),
|
|
created_by=entry_created_by,
|
|
source_trust=entry_source_trust,
|
|
sensitivity=sensitivity,
|
|
source_refs=source_refs,
|
|
run_id=run_id,
|
|
activity_id=activity_id,
|
|
decision_id=decision_id,
|
|
confirmed_by_user_action=confirmed_by_user_action,
|
|
reviewed_operation="promote" if adopted_by_user else None,
|
|
reviewed_memory_id=reviewed_source_memory_id,
|
|
)
|
|
|
|
def consolidate_scope(
|
|
self,
|
|
*,
|
|
scope_type: str,
|
|
scope_id: str,
|
|
reason: str,
|
|
request_id: str,
|
|
actor_type: Literal["user", "extractor"] = "extractor",
|
|
) -> MemoryConsolidationResult:
|
|
"""Safely remove exact duplicate, unreviewed machine-created items.
|
|
|
|
Consolidation is intentionally conservative: it does not summarize
|
|
History, semantically merge unrelated statements, or alter anything
|
|
the user created, confirmed, or pinned. More subjective changes remain
|
|
explicit user edits/HumanInteractions.
|
|
"""
|
|
|
|
if not request_id.strip():
|
|
raise ValueError("Memory consolidation request_id is required")
|
|
entries = self.list_entries(scope_type, scope_id)
|
|
groups: dict[tuple[str, str], list[MemoryEntryRecord]] = {}
|
|
for entry in entries:
|
|
key = (entry.kind, _normalized_memory_content(entry.content))
|
|
groups.setdefault(key, []).append(entry)
|
|
|
|
removed: list[str] = []
|
|
retained: list[str] = []
|
|
released = 0
|
|
for duplicates in groups.values():
|
|
if len(duplicates) < 2:
|
|
continue
|
|
ordered = sorted(duplicates, key=_memory_retention_key)
|
|
keeper = ordered[0]
|
|
retained.append(keeper.memory_id)
|
|
for candidate in ordered[1:]:
|
|
if (
|
|
candidate.created_by == "user"
|
|
or candidate.confirmed_by_user
|
|
or candidate.pinned_by_user
|
|
):
|
|
retained.append(candidate.memory_id)
|
|
continue
|
|
result = self._journal.apply_memory_mutation(
|
|
mutation_id=stable_memory_id(
|
|
"mutation", request_id, candidate.memory_id
|
|
),
|
|
idempotency_key=(
|
|
f"{request_id}:remove-exact-duplicate:"
|
|
f"{candidate.memory_id}"
|
|
),
|
|
operation="remove",
|
|
scope_type=scope_type,
|
|
scope_id=scope_id,
|
|
memory_id=candidate.memory_id,
|
|
expected_version=candidate.version,
|
|
actor_type=actor_type,
|
|
reason=(
|
|
f"{reason}; exact duplicate of {keeper.memory_id}"
|
|
),
|
|
source_refs=candidate.source_refs,
|
|
)
|
|
if result.entry is not None and result.entry.deleted_at:
|
|
removed.append(candidate.memory_id)
|
|
released += candidate.token_count
|
|
|
|
current = self.scope(scope_type, scope_id)
|
|
final_state = self._journal.record_memory_consolidation_result(
|
|
scope_type,
|
|
scope_id,
|
|
expected_revision=current.revision,
|
|
)
|
|
return MemoryConsolidationResult(
|
|
scope_state=final_state,
|
|
removed_memory_ids=tuple(removed),
|
|
retained_memory_ids=tuple(dict.fromkeys(retained)),
|
|
tokens_released=released,
|
|
)
|
|
|
|
def update_entry(
|
|
self,
|
|
*,
|
|
memory_id: str,
|
|
expected_version: int,
|
|
content: str,
|
|
kind: str,
|
|
actor_type: Literal["agent", "user", "extractor"],
|
|
reason: str,
|
|
request_id: str,
|
|
priority: str = "normal",
|
|
source_trust: str | None = None,
|
|
source_refs: tuple[str, ...] = (),
|
|
sensitivity: str = "normal",
|
|
actor_id: str | None = None,
|
|
run_id: str | None = None,
|
|
activity_id: str | None = None,
|
|
decision_id: str | None = None,
|
|
confirmed_by_user_action: bool = False,
|
|
) -> MemoryMutationResult:
|
|
existing = self._require_entry(memory_id)
|
|
# 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,
|
|
content=content,
|
|
actor_type=actor_type,
|
|
source_trust=trust,
|
|
confirmed_by_user_action=confirmed_by_user_action,
|
|
)
|
|
return self._journal.apply_memory_mutation(
|
|
mutation_id=stable_memory_id("mutation", request_id),
|
|
idempotency_key=request_id,
|
|
operation="replace",
|
|
scope_type=existing.scope_type,
|
|
scope_id=existing.scope_id,
|
|
memory_id=memory_id,
|
|
expected_version=expected_version,
|
|
actor_type=actor_type,
|
|
actor_id=actor_id,
|
|
reason=reason,
|
|
content=content,
|
|
kind=kind,
|
|
priority=priority,
|
|
token_count=count_tokens(content),
|
|
created_by=actor_type,
|
|
source_trust=trust,
|
|
sensitivity=sensitivity,
|
|
source_refs=source_refs,
|
|
run_id=run_id,
|
|
activity_id=activity_id,
|
|
decision_id=decision_id,
|
|
confirmed_by_user_action=confirmed_by_user_action,
|
|
)
|
|
|
|
def transition_entry(
|
|
self,
|
|
*,
|
|
memory_id: str,
|
|
expected_version: int,
|
|
operation: Literal["remove", "restore", "confirm", "pin"],
|
|
actor_type: Literal["agent", "user"],
|
|
reason: str,
|
|
request_id: str,
|
|
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 decision_id is None
|
|
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"}
|
|
)
|
|
):
|
|
raise PermissionError(
|
|
"Agent Memory changes outside unconfirmed Project entries "
|
|
"require a HumanInteraction"
|
|
)
|
|
return self._journal.apply_memory_mutation(
|
|
mutation_id=stable_memory_id("mutation", request_id),
|
|
idempotency_key=request_id,
|
|
operation=operation,
|
|
scope_type=existing.scope_type,
|
|
scope_id=existing.scope_id,
|
|
memory_id=memory_id,
|
|
expected_version=expected_version,
|
|
actor_type=actor_type,
|
|
actor_id=actor_id,
|
|
reason=reason,
|
|
run_id=run_id,
|
|
activity_id=activity_id,
|
|
decision_id=decision_id,
|
|
confirmed_by_user_action=decision_id is not None,
|
|
)
|
|
|
|
def search_memory(
|
|
self,
|
|
*,
|
|
project_id: str,
|
|
space_id: str | None = None,
|
|
user_id: str | None = None,
|
|
query: str = "",
|
|
token_budget: int = 2048,
|
|
) -> tuple[MemoryEntryRecord, ...]:
|
|
if token_budget < 1 or token_budget > 2048:
|
|
raise ValueError("Memory token budget must be between 1 and 2048")
|
|
candidates: list[MemoryEntryRecord] = []
|
|
scopes = [("project", project_id)]
|
|
if space_id:
|
|
scopes.append(("space", space_id))
|
|
if user_id:
|
|
scopes.append(("user", user_id))
|
|
query_tokens = {
|
|
token.casefold() for token in _QUERY_TOKEN_RE.findall(query)
|
|
}
|
|
for scope_type, scope_id in scopes:
|
|
state = self.scope(scope_type, scope_id)
|
|
if not state.use_enabled:
|
|
continue
|
|
candidates.extend(self.list_entries(scope_type, scope_id))
|
|
candidates.sort(
|
|
key=lambda entry: (
|
|
{"project": 0, "space": 1, "user": 2}[entry.scope_type],
|
|
not entry.pinned_by_user,
|
|
not entry.confirmed_by_user,
|
|
entry.priority != "high",
|
|
-entry.updated_at,
|
|
)
|
|
)
|
|
selected: list[MemoryEntryRecord] = []
|
|
used = 0
|
|
for entry in candidates:
|
|
if query_tokens and not (
|
|
query_tokens
|
|
& {
|
|
token.casefold()
|
|
for token in _QUERY_TOKEN_RE.findall(entry.content)
|
|
}
|
|
):
|
|
continue
|
|
if used + entry.token_count > token_budget:
|
|
continue
|
|
selected.append(entry)
|
|
used += entry.token_count
|
|
return tuple(selected)
|
|
|
|
def search_history(
|
|
self,
|
|
*,
|
|
project_id: str,
|
|
query: str = "",
|
|
after_cursor: str | None = None,
|
|
limit: int = 50,
|
|
byte_budget: int = 64 * 1024,
|
|
token_budget: int = 4096,
|
|
) -> HistoryQueryPage:
|
|
cursor = parse_project_cursor(after_cursor)
|
|
if limit < 1 or limit > 100:
|
|
raise ValueError("History limit must be between 1 and 100")
|
|
if byte_budget < 256 or byte_budget > 512 * 1024:
|
|
raise ValueError("History byte budget is out of bounds")
|
|
if token_budget < 64 or token_budget > 16384:
|
|
raise ValueError("History token budget is out of bounds")
|
|
query_text = query.casefold().strip()
|
|
scan_cursor = cursor
|
|
selected: list[HistoryQueryResult] = []
|
|
used_bytes = 0
|
|
used_tokens = 0
|
|
source_watermark = self._journal.get_project_history_cursor(project_id)
|
|
exhausted = False
|
|
while len(selected) < limit and scan_cursor < source_watermark:
|
|
page = self._journal.list_project_history_events(
|
|
project_id,
|
|
after_cursor=scan_cursor,
|
|
limit=min(100, max(limit * 2, 20)),
|
|
)
|
|
if not page:
|
|
exhausted = True
|
|
break
|
|
for item in page:
|
|
previous_cursor = scan_cursor
|
|
scan_cursor = item.journal_cursor
|
|
payload = _history_payload_projection(
|
|
item.event.event_type, item.event.payload
|
|
)
|
|
searchable = json.dumps(payload, ensure_ascii=False).casefold()
|
|
if (
|
|
query_text
|
|
and query_text not in searchable
|
|
and query_text not in item.event.event_type.casefold()
|
|
):
|
|
continue
|
|
encoded = json.dumps(
|
|
payload, ensure_ascii=False, separators=(",", ":")
|
|
)
|
|
item_bytes = len(encoded.encode("utf-8"))
|
|
item_tokens = count_tokens(encoded)
|
|
if (
|
|
used_bytes + item_bytes > byte_budget
|
|
or used_tokens + item_tokens > token_budget
|
|
):
|
|
# The unreturned item remains addressable by the next page;
|
|
# never advance a durable cursor past evidence the caller
|
|
# did not receive.
|
|
scan_cursor = previous_cursor
|
|
exhausted = False
|
|
break
|
|
selected.append(
|
|
HistoryQueryResult(
|
|
citation_id=(
|
|
f"history:{project_id}:{item.journal_cursor}"
|
|
),
|
|
journal_cursor=item.journal_cursor,
|
|
event_id=item.event.event_id,
|
|
run_id=item.event.run_id,
|
|
event_type=item.event.event_type,
|
|
content=payload,
|
|
source_trust=event_source_trust(item.event.event_type),
|
|
created_at=item.event.created_at,
|
|
)
|
|
)
|
|
used_bytes += item_bytes
|
|
used_tokens += item_tokens
|
|
if len(selected) >= limit:
|
|
break
|
|
else:
|
|
if scan_cursor >= source_watermark:
|
|
exhausted = True
|
|
continue
|
|
break
|
|
complete = exhausted or scan_cursor >= source_watermark
|
|
next_cursor = format_project_cursor(scan_cursor)
|
|
current = format_project_cursor(source_watermark)
|
|
return HistoryQueryPage(
|
|
source="sqlite",
|
|
project_id=project_id,
|
|
complete=complete,
|
|
freshness="current",
|
|
watermark_kind="journal_cursor",
|
|
source_watermark=current,
|
|
available_through_watermark=current,
|
|
items=tuple(selected),
|
|
next_cursor=next_cursor,
|
|
redactions=(
|
|
"credential_keys_and_recognized_values",
|
|
"raw_connector_tool_results",
|
|
"local_artifact_paths_and_content",
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _assert_mutation_policy(
|
|
*,
|
|
scope_type: str,
|
|
kind: str,
|
|
content: str,
|
|
actor_type: str,
|
|
source_trust: str,
|
|
confirmed_by_user_action: bool = False,
|
|
entry_created_by: str | None = None,
|
|
) -> None:
|
|
assert_memory_entry_policy(
|
|
kind=kind,
|
|
content=content,
|
|
created_by=entry_created_by or actor_type,
|
|
source_trust=source_trust,
|
|
confirmed_by_user=(
|
|
actor_type == "user" or confirmed_by_user_action
|
|
),
|
|
)
|
|
if (
|
|
actor_type == "agent"
|
|
and scope_type != "project"
|
|
and not confirmed_by_user_action
|
|
):
|
|
raise PermissionError(
|
|
"Agent Space/User Memory mutations require HumanInteraction"
|
|
)
|
|
|
|
def _require_entry(self, memory_id: str) -> MemoryEntryRecord:
|
|
entry = self._journal.get_memory_entry(memory_id)
|
|
if entry is None:
|
|
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"):
|
|
return "user_asserted"
|
|
if event_type.startswith("tool."):
|
|
return "tool_observed"
|
|
if event_type.startswith(("run.", "approval.", "human_interaction.")):
|
|
return "system_verified"
|
|
if event_type == "assistant.final" or event_type.startswith("assistant."):
|
|
return "model_inferred"
|
|
if event_type.startswith("legacy."):
|
|
return "legacy_unverified"
|
|
return "system_verified"
|
|
|
|
|
|
def stable_memory_id(namespace: str, *parts: str) -> str:
|
|
digest = hashlib.sha256(
|
|
json.dumps(
|
|
[namespace, *parts],
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
).hexdigest()
|
|
prefix = "mut" if namespace == "mutation" else "mem"
|
|
return f"{prefix}_{digest[:32]}"
|
|
|
|
|
|
def _normalized_memory_content(value: str) -> str:
|
|
return " ".join(unicodedata.normalize("NFKC", value).split()).casefold()
|
|
|
|
|
|
def _memory_retention_key(entry: MemoryEntryRecord) -> tuple:
|
|
trust_rank = {
|
|
"user_confirmed": 0,
|
|
"user_asserted": 1,
|
|
"system_verified": 2,
|
|
"tool_observed": 3,
|
|
"model_inferred": 4,
|
|
"external_untrusted": 5,
|
|
"legacy_unverified": 6,
|
|
}
|
|
return (
|
|
not entry.pinned_by_user,
|
|
not entry.confirmed_by_user,
|
|
entry.created_by != "user",
|
|
entry.priority != "high",
|
|
trust_rank.get(entry.source_trust, 99),
|
|
-entry.updated_at,
|
|
entry.memory_id,
|
|
)
|
|
|
|
|
|
def format_project_cursor(value: int) -> str:
|
|
return f"sqlite-project-v1:{value}"
|
|
|
|
|
|
def parse_project_cursor(value: str | None) -> int:
|
|
if value is None or value == "":
|
|
return 0
|
|
prefix = "sqlite-project-v1:"
|
|
if not value.startswith(prefix):
|
|
raise ValueError("unsupported Project History cursor")
|
|
try:
|
|
cursor = int(value.removeprefix(prefix))
|
|
except ValueError as exc:
|
|
raise ValueError("invalid Project History cursor") from exc
|
|
if cursor < 0:
|
|
raise ValueError("invalid Project History cursor")
|
|
return cursor
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _token_encoding():
|
|
if tiktoken is None:
|
|
return None
|
|
try:
|
|
# Some tiktoken builds lazily download their vocabulary. Memory must
|
|
# remain available offline, so a missing cache falls back once to the
|
|
# conservative local estimate instead of creating a network loop.
|
|
return tiktoken.get_encoding("cl100k_base")
|
|
except Exception: # noqa: BLE001 - optional offline optimization
|
|
return None
|
|
|
|
|
|
def count_tokens(value: str) -> int:
|
|
encoding = _token_encoding()
|
|
if encoding is None:
|
|
return max(1, (len(value) + 2) // 3)
|
|
return max(1, len(encoding.encode(value)))
|
|
|
|
|
|
def get_lightweight_memory_service() -> LightweightMemoryService:
|
|
return LightweightMemoryService(get_default_run_journal())
|