Files
pocketpaw/tests/test_identity_api.py
Rohit Kushwaha 1b4d512c24 fix(agents): prevent identity drift in long conversations (#554)
* fix(bootstrap): move identity block after instructions, wrap in <identity> XML tags

Addresses issue #131 (identity gets diluted by 100+ lines of tool docs).

- Tool instructions are now placed FIRST in the system prompt so they act
  as background reference material
- Identity/soul/style/user_profile are placed LAST, wrapped in
  <identity>...</identity> XML tags, keeping them closest to the live
  conversation turns where the model pays the most attention
- User profile (USER.md) is now also inside the <identity> block

* fix(agents): add shared _DEFAULT_IDENTITY constant to backend.py

Single source of truth for the system-prompt fallback used by all backends
when no identity context is provided. All backends now import this constant
instead of using their own ad-hoc minimal inline strings.

* fix(agents): use _DEFAULT_IDENTITY fallback across all backends

Fixes identity loss when no system_prompt is provided (issue #131 pt.3).

- codex_cli: always inject effective_system (was silently omitted when
  system_prompt was falsy)
- opencode: replace empty-string fallback with _DEFAULT_IDENTITY
- copilot_sdk: always inject identity in prompt_parts and session_opts
  system_message (was conditionally skipped)
- openai_agents: replace inline 'You are PocketPaw...' string with
  _DEFAULT_IDENTITY
- google_adk: same replacement as openai_agents

* fix(loop): add periodic identity reinforcement for long conversations

Adds _IDENTITY_REINFORCE_THRESHOLD = 20. When the session history reaches
this length the agent loop appends a compact '# Identity Reminder' block to
the system prompt, nudging the model back on-character without a full
re-injection that would waste context window (issue #131 pt.4).

* test: add tests for identity drift fixes (#131)

- test_bootstrap.py: add 4 new assertions for the restructured
  to_system_prompt() — XML tags present, instructions before identity,
  user_profile inside <identity> block, prompt starts with <identity>
  when no instructions are provided
- test_backend_protocol.py: add TestDefaultIdentity verifying
  _DEFAULT_IDENTITY is a non-empty string that mentions PocketPaw
- test_agent_loop.py: add two new async tests —
  * identity reinforcement IS appended when history >= threshold
  * identity reinforcement is NOT appended for short conversations

* style: apply ruff-format fixes flagged by pre-commit

Formatting corrections auto-applied by ruff-format hook:
- tests/test_agent_loop.py: wrap long AsyncMock lines (E501)
- tests/test_backend_protocol.py: minor whitespace
- src/pocketpaw/agents/claude_sdk.py: trailing whitespace

* fix(tests): update test_instructions_between_style_and_knowledge for new prompt layout

PR #548 restructured to_system_prompt() so tool instructions come first
and the identity block (style, soul, user_profile) is wrapped in <identity>
tags at the end. The assertion in test_identity_api.py still expected the
old ordering (style < instructions < user). Update it to match the new
layout: instructions < style < user.

---------

Co-authored-by: Ragini Pandey <pandeyragini55@gmail.com>
Co-authored-by: Ragini Pandey <99394366+ragini-pandey@users.noreply.github.com>
2026-03-10 21:06:25 +05:30

340 lines
13 KiB
Python

"""Tests for Identity API — GET + PUT /api/identity.
Covers:
- GET /api/identity returns all 5 identity files (including instructions)
- PUT /api/identity saves edits to disk
- PUT /api/identity partial update (only some files)
- Agent picks up file changes on next prompt build
Created: 2026-02-12
Updated: 2026-02-18 — Added instructions_file coverage
"""
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from pocketpaw.bootstrap.default_provider import DefaultBootstrapProvider
class TestGetIdentity:
"""Tests for GET /api/identity."""
async def test_returns_all_five_files(self):
"""GET /api/identity returns identity, soul, style, instructions, and user_file."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
# Write known content
(base / "IDENTITY.md").write_text("I am PocketPaw")
(base / "SOUL.md").write_text("I value privacy")
(base / "STYLE.md").write_text("Be concise")
(base / "INSTRUCTIONS.md").write_text("Be agentic")
(base / "USER.md").write_text("Name: Alice")
with (
patch("pocketpaw.dashboard.get_config_path") as mock_path,
patch(
"pocketpaw.dashboard.DefaultBootstrapProvider",
return_value=provider,
),
):
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import get_identity
result = await get_identity()
assert result["identity_file"] == "I am PocketPaw"
assert result["soul_file"] == "I value privacy"
assert result["style_file"] == "Be concise"
assert result["instructions_file"] == "Be agentic"
assert result["user_file"] == "Name: Alice"
async def test_returns_default_user_profile(self):
"""GET /api/identity returns default USER.md when not customized."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
with (
patch("pocketpaw.dashboard.get_config_path") as mock_path,
patch(
"pocketpaw.dashboard.DefaultBootstrapProvider",
return_value=provider,
),
):
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import get_identity
result = await get_identity()
assert "user_file" in result
assert "# User Profile" in result["user_file"]
async def test_returns_default_instructions(self):
"""GET /api/identity returns default INSTRUCTIONS.md when not customized."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
with (
patch("pocketpaw.dashboard.get_config_path") as mock_path,
patch(
"pocketpaw.dashboard.DefaultBootstrapProvider",
return_value=provider,
),
):
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import get_identity
result = await get_identity()
assert "instructions_file" in result
assert "PocketPaw Tools" in result["instructions_file"]
assert "Guidelines" in result["instructions_file"]
class TestSaveIdentity:
"""Tests for PUT /api/identity."""
async def test_saves_all_files(self):
"""PUT /api/identity writes all 5 files to disk."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
identity_dir = base / "identity"
identity_dir.mkdir()
request = MagicMock()
request.json = AsyncMock(
return_value={
"identity_file": "New identity",
"soul_file": "New soul",
"style_file": "New style",
"instructions_file": "New instructions",
"user_file": "Name: Bob\nTimezone: EST",
}
)
with patch("pocketpaw.dashboard.get_config_path") as mock_path:
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import save_identity
result = await save_identity(request)
assert result["ok"] is True
assert set(result["updated"]) == {
"IDENTITY.md",
"SOUL.md",
"STYLE.md",
"INSTRUCTIONS.md",
"USER.md",
}
assert (identity_dir / "IDENTITY.md").read_text(encoding="utf-8") == "New identity"
assert (identity_dir / "SOUL.md").read_text(encoding="utf-8") == "New soul"
assert (identity_dir / "STYLE.md").read_text(encoding="utf-8") == "New style"
assert (identity_dir / "INSTRUCTIONS.md").read_text(
encoding="utf-8"
) == "New instructions"
assert (identity_dir / "USER.md").read_text(
encoding="utf-8"
) == "Name: Bob\nTimezone: EST"
async def test_partial_update(self):
"""PUT /api/identity with only user_file updates only that file."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
identity_dir = base / "identity"
identity_dir.mkdir()
(identity_dir / "IDENTITY.md").write_text("Original identity")
(identity_dir / "USER.md").write_text("Name: Old")
request = MagicMock()
request.json = AsyncMock(
return_value={
"user_file": "Name: Updated",
}
)
with patch("pocketpaw.dashboard.get_config_path") as mock_path:
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import save_identity
result = await save_identity(request)
assert result["ok"] is True
assert result["updated"] == ["USER.md"]
# Original identity untouched
assert (identity_dir / "IDENTITY.md").read_text(encoding="utf-8") == "Original identity"
# User file updated
assert (identity_dir / "USER.md").read_text(encoding="utf-8") == "Name: Updated"
async def test_ignores_non_string_values(self):
"""PUT /api/identity ignores non-string values."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
identity_dir = base / "identity"
identity_dir.mkdir()
request = MagicMock()
request.json = AsyncMock(
return_value={
"identity_file": 42, # not a string
"soul_file": "Valid soul",
}
)
with patch("pocketpaw.dashboard.get_config_path") as mock_path:
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import save_identity
result = await save_identity(request)
assert result["ok"] is True
assert result["updated"] == ["SOUL.md"]
async def test_creates_identity_dir_if_missing(self):
"""PUT /api/identity creates the identity/ directory if it doesn't exist."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
request = MagicMock()
request.json = AsyncMock(return_value={"user_file": "Name: New User"})
with patch("pocketpaw.dashboard.get_config_path") as mock_path:
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import save_identity
result = await save_identity(request)
assert result["ok"] is True
assert (base / "identity" / "USER.md").read_text(encoding="utf-8") == "Name: New User"
async def test_ignores_unknown_keys(self):
"""PUT /api/identity ignores keys not in the file_map."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
identity_dir = base / "identity"
identity_dir.mkdir()
request = MagicMock()
request.json = AsyncMock(
return_value={
"user_file": "Name: Valid",
"malicious_key": "should be ignored",
}
)
with patch("pocketpaw.dashboard.get_config_path") as mock_path:
mock_path.return_value = base / "config.json"
from pocketpaw.dashboard import save_identity
result = await save_identity(request)
assert result["updated"] == ["USER.md"]
assert not (identity_dir / "malicious_key").exists()
async def test_invalid_json_returns_400(self):
request = MagicMock()
request.json = AsyncMock(side_effect=ValueError("Invalid JSON"))
from fastapi import HTTPException
from pocketpaw.dashboard import save_identity
with pytest.raises(HTTPException) as exc_info:
await save_identity(request)
assert exc_info.value.status_code == 400
class TestIdentityAgentIntegration:
"""Tests verifying that saved identity changes are picked up by the agent."""
async def test_saved_user_profile_in_system_prompt(self):
"""After saving USER.md, the next get_context() picks it up."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
# Initially has default content
ctx = await provider.get_context()
assert "(your name)" in ctx.user_profile
# Simulate saving via API (write directly)
(base / "USER.md").write_text("Name: Charlie\nTimezone: UTC+5")
# Next call picks up the change
ctx2 = await provider.get_context()
assert ctx2.user_profile == "Name: Charlie\nTimezone: UTC+5"
assert "Name: Charlie" in ctx2.to_system_prompt()
async def test_saved_identity_in_system_prompt(self):
"""After saving IDENTITY.md, the next get_context() picks it up."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
(base / "IDENTITY.md").write_text("I am a custom agent named Luna.")
ctx = await provider.get_context()
assert ctx.identity == "I am a custom agent named Luna."
assert "Luna" in ctx.to_system_prompt()
async def test_all_files_in_system_prompt(self):
"""All 5 identity files appear in the system prompt."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
(base / "IDENTITY.md").write_text("CUSTOM_IDENTITY")
(base / "SOUL.md").write_text("CUSTOM_SOUL")
(base / "STYLE.md").write_text("CUSTOM_STYLE")
(base / "INSTRUCTIONS.md").write_text("CUSTOM_INSTRUCTIONS")
(base / "USER.md").write_text("CUSTOM_USER")
ctx = await provider.get_context()
prompt = ctx.to_system_prompt()
assert "CUSTOM_IDENTITY" in prompt
assert "CUSTOM_SOUL" in prompt
assert "CUSTOM_STYLE" in prompt
assert "CUSTOM_INSTRUCTIONS" in prompt
assert "CUSTOM_USER" in prompt
async def test_instructions_between_style_and_knowledge(self):
"""Instructions (tool docs) appear before the identity block; user profile is
inside the identity block, after style."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
(base / "INSTRUCTIONS.md").write_text("INSTR_MARKER")
(base / "STYLE.md").write_text("STYLE_MARKER")
(base / "USER.md").write_text("USER_MARKER")
ctx = await provider.get_context()
prompt = ctx.to_system_prompt()
style_pos = prompt.index("STYLE_MARKER")
instr_pos = prompt.index("INSTR_MARKER")
user_pos = prompt.index("USER_MARKER")
# New layout: instructions first, then <identity> block (style … user_profile)
assert instr_pos < style_pos < user_pos
async def test_saved_instructions_in_system_prompt(self):
"""After saving INSTRUCTIONS.md, the next get_context() picks it up."""
with tempfile.TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
provider = DefaultBootstrapProvider(base_path=base)
# Initially has default content
ctx = await provider.get_context()
assert "PocketPaw Tools" in ctx.instructions
# Simulate saving via API (write directly)
(base / "INSTRUCTIONS.md").write_text("Custom tool instructions here")
# Next call picks up the change
ctx2 = await provider.get_context()
assert ctx2.instructions == "Custom tool instructions here"
assert "Custom tool instructions here" in ctx2.to_system_prompt()