feat(memory): agent-scope two-phase trajectory/experience memory pipeline

[yangxinxin-7]

Overview

Introduces an agent-scope memory pipeline that runs alongside the existing user memory extraction. After each session is committed, two sequential phases extract and consolidate execution knowledge specific to the agent identity.

Architecture

Two-phase pipeline

Phase 1 — Trajectory extraction
Reads the full conversation and writes one (or rarely more) trajectory memory per business domain. Each trajectory is an immutable execution trace: goal, step-by-step actions with tool calls, result, and failure analysis. Filenames are timestamped to guarantee uniqueness (<name>_YYYYMMDDHHMMSS.md).

Phase 2 — Experience consolidation
For each newly written trajectory, retrieves the top-K most relevant existing experience memories (via vector search with directory-listing fallback), then lets the LLM choose one of four strategies:

• Update — trajectory fits an existing experience; rewrite it in place.
• Replace — related experience exists but its name no longer captures the broader pattern; create a new one and delete the old.
• Create — no related experience exists; create a new one.
• Skip — no transferable lesson; do nothing.

Each experience file stores a source_trajectories metadata list (capped at 5 entries) so Phase 2 can ground its decision in concrete past executions.

Isolation from user memory

New agent_only: true flag on MemoryTypeSchema. Schemas marked agent_only (trajectory, experience) are filtered out of SessionExtractContextProvider so they never appear in user-scope extraction. User memory and agent memory run concurrently via asyncio.gather.

Changes

Area	Change
memory/experience.yaml	New schema replacing cases.yaml
memory/trajectory.yaml	New schema replacing patterns.yaml
core/directories.py	Rename preset dirs: cases→trajectories, patterns→experiences
compressor_v2.py	extract_agent_memories(), _run_extract_phase(), _append_trajectories_to_experiences()
agent_trajectory_context_provider.py	Phase 1 provider (no prefetch, add_only)
agent_experience_context_provider.py	Phase 2 provider (search + prefetch candidates + source trajectories)
session.py	Concurrent gather with independent error handling per task
memory_config.py	agent_memory_enabled flag (default false)
client/local.py	Accept explicit UserIdentifier for embedded/test use
extract_loop.py	Track prefetched URIs in _read_files; guard empty tool schema

Configuration

Enable via ~/.openviking/ov.conf:

{
  "memory": {
    "version": "v2",
    "agent_memory_enabled": true
  }
}

Testing

End-to-end integration test in tests/integration/test_agent_memory_e2e.py:

• Two sessions in the same booking-conflict domain
• Round 1: asserts trajectory written + experience created
• Round 2: asserts trajectory count grows + experience count stays at 1 (edit path, not duplicate create)
• Verifies source_trajectories metadata links back to extracted trajectories

Run:

RUN_AGENT_MEMORY_TESTS=1 .venv/bin/pytest tests/integration/test_agent_memory_e2e.py -v -s -m integration

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

🏅 Score: 75

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Race condition in _append_trajectories_to_experiences _append_trajectories_to_experiences reads and writes experience files after the lock on the experience directory has been released. This can lead to lost updates if multiple agent memory extraction runs occur concurrently. async def _append_trajectories_to_experiences( self, exp_uris: List[str], traj_uris: List[str], ctx, viking_fs, ) -> None: """Append traj_uris to the source_trajectories list of each experience file. This is the system-side management of source_trajectories — the LLM never outputs this field; the pipeline appends the batch after a write or edit. """ from openviking.session.memory.utils.content import deserialize_full, serialize_with_metadata normalized_traj_uris = [uri for uri in traj_uris if uri] if not normalized_traj_uris: return for exp_uri in exp_uris: try: raw = await viking_fs.read_file(exp_uri, ctx=ctx) or "" file_content = deserialize_full(raw) plain_content = file_content.plain_content metadata = file_content.memory_fields or {} existing = metadata.get("source_trajectories", []) if isinstance(existing, list): uris = list(existing) elif isinstance(existing, str) and existing.strip(): uris = [line.strip() for line in existing.splitlines() if line.strip()] else: uris = [] changed = False for traj_uri in normalized_traj_uris: if traj_uri not in uris: uris.append(traj_uri) changed = True # Trim to the most recent N entries so the list doesn't grow unboundedly. if len(uris) > MAX_SOURCE_TRAJECTORIES: uris = uris[-MAX_SOURCE_TRAJECTORIES:] changed = True if changed: metadata["source_trajectories"] = uris metadata["content"] = plain_content new_raw = serialize_with_metadata(metadata) await viking_fs.write_file(exp_uri, new_raw, ctx=ctx) tracer.info( f"[source_traj] appended {len(normalized_traj_uris)} trajectories -> {exp_uri}" ) else: tracer.info(f"[source_traj] already present, skip: {exp_uri}") except Exception as e: logger.warning(f"Failed to append source trajectories to {exp_uri}: {e}") Malformed experience files due to missing content parameter in serialize_with_metadata In _append_trajectories_to_experiences, serialize_with_metadata is called without passing the content parameter, and content is instead set in metadata. This leads to content being stored in the YAML frontmatter instead of the file body, and an empty body. metadata["source_trajectories"] = uris metadata["content"] = plain_content new_raw = serialize_with_metadata(metadata) await viking_fs.write_file(exp_uri, new_raw, ctx=ctx)

[github-actions]

PR Code Suggestions ✨

No code suggestions found for the PR.