Day 6：Session + Memory（三层记忆）

session JSONL 负责续聊；system prompt 负责规则和索引；memdir 负责长期事实。

class Session:
    """一次会话。管理 session id、JSONL 落盘、读取历史消息。"""

    @classmethod
    def create(cls, cwd: Path) -> "Session":
        sid = uuid.uuid4().hex[:12]
        file_path = _sessions_dir(cwd) / f"{sid}.jsonl"
        file_path.touch()
        return cls(cwd=cwd, session_id=sid, file_path=file_path, resumed=False)

    @classmethod
    def load_latest(cls, cwd: Path) -> "Session | None":
        sessions_dir = _sessions_dir(cwd)
        jsonl_files = list(sessions_dir.glob("*.jsonl"))
        if not jsonl_files:
            return None
        latest = max(jsonl_files, key=lambda p: p.stat().st_mtime)
        return cls(cwd=cwd, session_id=latest.stem, file_path=latest, resumed=True)

    def append_messages(self, msgs: list[dict[str, Any]]) -> None:
        now = datetime.now(timezone.utc).isoformat()
        with open(self.file_path, "a", encoding="utf-8") as f:
            for msg in msgs:
                record = {"role": msg["role"], "content": msg["content"], "timestamp": now}
                f.write(json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n")

if session and session.history:
    messages = list(session.history)
    messages.append({"role": "user", "content": prompt})
else:
    messages = [{"role": "user", "content": prompt}]

if session:
    session.append_messages([messages[-1]])

$ uv run agent-code "用 echo 工具说 hello"
Agent Code
cwd: /your/project
provider: anthropic  model: deepseek-v4-flash
session: a1b2c3d4e5f6

tool_call: echo {'text': 'hello'}
observation: hello
final: 已经用 echo 工具说出了 "hello"。

$ uv run agent-code -c "我们上次说了什么"
Agent Code
cwd: /your/project
provider: anthropic  model: deepseek-v4-flash
session: a1b2c3d4e5f6 (resumed)

final: 我们上次用 echo 工具说了 "hello"。

def load_agent_md(cwd: Path) -> str | None:
    """读取 cwd 下的 AGENT.md，包装成 <project-rules> 块。
    文件不存在返回 None——不是错误，只是没配置。"""
    agent_md = cwd / "AGENT.md"
    if not agent_md.exists():
        return None
    content = agent_md.read_text(encoding="utf-8", errors="replace").strip()
    if not content:
        return None
    if len(content.encode("utf-8")) > _MAX_AGENT_MD_BYTES:
        truncated = content.encode("utf-8")[:_MAX_AGENT_MD_BYTES].decode("utf-8", errors="replace")
        content = truncated + "\n\n[... AGENT.md truncated at 50 KB ...]"
    return f"<project-rules>\n{content}\n</project-rules>"

class ModelProvider(Protocol):
    def complete(
        self,
        messages: list[dict[str, Any]],
        tools: list[Any] | None = None,
        system: str | None = None,
    ) -> ModelResponse:
        ...

_SYSTEM_CORE = (
    "You are an AI coding agent running inside a CLI harness. "
    "You have access to tools for reading/writing files, running shell commands, "
    "searching the web, and asking the user questions. "
    "Use tools when needed; respond directly when you can."
)


def build_system_prompt(cwd: Path) -> str:
    parts: list[str] = [_SYSTEM_CORE]
    agent_md = load_agent_md(cwd)
    if agent_md:
        parts.append(agent_md)
    return "\n\n".join(parts)

$ cat > AGENT.md << 'EOF'
# 项目规则（demo）
- 所有代码必须用 Python 3.10+，写完整 type hints。
- 测试框架用 pytest，测试文件放在 tests/ 目录下。
- 不要用 print 打日志，用 logging 模块。
EOF

$ uv run python -c "
from pathlib import Path
from agent_code.agent import build_system_prompt
print(build_system_prompt(Path.cwd()))
"

You are an AI coding agent running inside a CLI harness. ...

<project-rules>
# 项目规则（demo）
- 所有代码必须用 Python 3.10+，写完整 type hints。
- 测试框架用 pytest，测试文件放在 tests/ 目录下。
- 不要用 print 打日志，用 logging 模块。
</project-rules>

def compact(messages: list[dict[str, Any]], keep: int = 8) -> list[dict[str, Any]]:
    """确定性压缩消息历史。不调 LLM。"""
    pin_count = 2
    if len(messages) <= keep + pin_count:
        return messages

    pinned = messages[:pin_count]
    working = messages[-keep:]
    middle = messages[pin_count:-keep]
    compressed = _build_compressed_block(middle)
    return pinned + [compressed] + working

if len(messages) > 40:
    messages = compact(messages, keep=8)
    console.print(f"[dim]compacted: {len(messages)} messages remaining[/dim]")

$ uv run python -c "
from agent_code.compact_basic import compact

msgs = []
for i in range(15):
    msgs.append({'role': 'user', 'content': f'task step {i}'})
    msgs.append({'role': 'assistant', 'content': [
        {'type': 'text', 'text': f'doing step {i}'},
        {'type': 'tool_use', 'id': f'c{i}', 'name': 'read_file', 'input': {'path': f'file{i}.py'}},
    ]})
    msgs.append({'role': 'user', 'content': [
        {'type': 'tool_result', 'tool_use_id': f'c{i}', 'content': f'content of file{i}'}
    ]})

print(f'before: {len(msgs)} messages')
result = compact(msgs, keep=8)
print(f'after: {len(result)} messages')
"

MEMORY_TYPES = ("user", "feedback", "project", "reference")


@dataclass
class MemoryEntry:
    mem_type: str
    title: str
    slug: str
    body: str
    file_path: str

_LOW_RISK_WRITES = frozenset({"memory_write"})

if tool_name in _LOW_RISK_WRITES:
    return PermissionDecision("allow")

def build_system_prompt(cwd: Path) -> str:
    """组装 system prompt：核心指南 + AGENT.md + MEMORY.md 索引。"""
    from .memdir.store import load_index as load_memory_index

    parts: list[str] = [_SYSTEM_CORE]
    agent_md = load_agent_md(cwd)
    if agent_md:
        parts.append(agent_md)

    memory_index = load_memory_index(cwd)
    if memory_index:
        parts.append(f"<project-memory>\n{memory_index}\n</project-memory>")

    return "\n\n".join(parts)

$ uv run python -c "
import tempfile
from pathlib import Path
from agent_code.memdir.store import write_memory, recall_memory
from agent_code.agent import build_system_prompt

with tempfile.TemporaryDirectory() as tmp:
    cwd = Path(tmp)
    entry = write_memory(cwd, 'user', '用户角色', '用户是数据科学家，主要研究方向是观测性。')
    print('written:', entry.file_path)
    hits = recall_memory(cwd, '数据科学家')
    print('recalled:', len(hits), 'entries')
    print('first hit type:', hits[0].mem_type, 'title:', hits[0].title)
    sp = build_system_prompt(cwd)
    print('project-memory in system prompt:', '<project-memory>' in sp)
"

written: .agent/memory/user/mem-247c9e11.md
recalled: 1 entries
first hit type: user title: 用户角色
project-memory in system prompt: True

$ uv run agent-code "我是数据科学家，主要研究观测性。记住这一点。"
tool_call: memory_write {'type': 'user', 'title': '用户角色', 'body': '用户是数据科学家，主要研究方向是观测性。'}
observation: Memory saved: [user] 用户角色 -> .agent/memory/user/mem-247c9e11.md
final: 已记住。下次对话时我会记得你是数据科学家，研究方向是观测性。

$ uv run agent-code "用 memory_recall 工具查一下你对我有什么记忆"
tool_call: memory_recall {'query': '用户 数据科学家 观测性', 'top_k': 5}
observation: ## [user] 用户角色
  file: .agent/memory/user/mem-247c9e11.md
  用户是数据科学家，主要研究方向是观测性。

final: 根据我的记忆，你是一名数据科学家，主要研究观测性方向。

$ uv run agent-code "hello"
$ uv run agent-code -c "继续"