メモリストアの棚卸しが半分しか返らない — path_prefix セグメント一致への移行

# 変更前。これで /projects/alpha/... も /projects/alpha_archive/... も拾えていた
memories = client.beta.memory_stores.memories.list(
    memory_store_id=store_id,
    path_prefix="/projects/alpha",   # ← 末尾スラッシュが無い
    view="basic",
    betas=["managed-agents-2026-04-01"],
)

from anthropic import Anthropic
 
client = Anthropic()  # ANTHROPIC_API_KEY は環境変数から
BETAS = ["managed-agents-2026-04-01", "agent-memory-2026-07-22"]
 
 
def normalize_prefix(prefix: str) -> str:
    """path_prefix は必ず末尾スラッシュで終える。ルートは "/" に統一。"""
    if not prefix.startswith("/"):
        prefix = "/" + prefix
    if not prefix.endswith("/"):
        prefix = prefix + "/"
    return prefix
 
 
def list_all_memories(store_id: str, prefix: str = "/") -> list[dict]:
    """
    prefix 配下のメモリを全件返す。
    - 返却順に依存しない（呼び出し側で必ずソートする）
    - depth は 0/1/省略のみ許されるため、木は自分で再帰的に歩く
    """
    prefix = normalize_prefix(prefix)
    collected: list[dict] = []
    cursor: str | None = None
 
    while True:
        page = client.beta.memory_stores.memories.list(
            memory_store_id=store_id,
            path_prefix=prefix,
            depth=1,          # 直下の子だけを取り、サブツリーは再帰で辿る
            view="basic",     # メタデータのみ。本文は必要な時だけ retrieve
            after_id=cursor,
            betas=BETAS,
        )
        for item in page.data:
            if item.type == "memory_prefix":
                # ディレクトリ相当のノード。1 段掘り下げる
                collected.extend(list_all_memories(store_id, item.path))
            else:
                collected.append({"id": item.id, "path": item.path})
 
        if not getattr(page, "has_more", False):
            break
        cursor = page.last_id
 
    # サーバ順は保証されないので、比較・差分のために自前で安定ソートする
    collected.sort(key=lambda m: m["path"])
    return collected

def list_project_including_archive(store_id: str, project: str) -> list[dict]:
    """プロジェクト本体とアーカイブを、意図的に別プレフィックスとして合算する。"""
    live = list_all_memories(store_id, f"/projects/{project}/")
    archive = list_all_memories(store_id, f"/projects/{project}_archive/")
    return live + archive

def safe_correct(store_id: str, memory_id: str, new_content: str) -> bool:
    current = client.beta.memory_stores.memories.retrieve(
        memory_store_id=store_id, memory_id=memory_id, betas=BETAS,
    )
    try:
        client.beta.memory_stores.memories.update(
            memory_store_id=store_id,
            memory_id=memory_id,
            content=new_content,
            precondition={
                "type": "content_sha256",
                "content_sha256": current.content_sha256,
            },
            betas=BETAS,
        )
        return True
    except Exception:
        # ハッシュ不一致 = 誰か（多くはエージェント自身）が先に書いた。
        # 黙って上書きせず、再読込して判断し直す
        return False

"""prefix_audit.py — 旧 substring 一致と新 segment 一致の差を、API を叩かずに数える。"""
import json, sys
 
 
def substring_match(paths, prefix):          # 変更前の挙動
    return [p for p in paths if p.startswith(prefix)]
 
 
def segment_match(paths, prefix):            # 変更後の挙動
    if not prefix.endswith("/"):
        return []                            # 末尾スラッシュ無しは一致しない
    return [p for p in paths if p.startswith(prefix)]
 
 
def prefix_nodes(paths):
    """depth=1 で再帰する際に踏む、ディレクトリ相当ノードの数。"""
    nodes = set()
    for p in paths:
        parts = p.strip("/").split("/")[:-1]
        for i in range(len(parts)):
            nodes.add("/" + "/".join(parts[: i + 1]) + "/")
    return nodes
 
 
if __name__ == "__main__":
    paths = json.load(open(sys.argv[1]))     # memories.list から集めた path 一覧
    print(f"total memories: {len(paths)}")
    n = len(prefix_nodes(paths))
    print(f"prefix nodes  : {n}  -> depth=1 recursion needs {n + 1} list calls\n")
    print(f"{'path_prefix':<28}{'substring':>10}{'as-written':>12}{'normalized':>12}{'lost':>7}")
    for pf in sys.argv[2:]:
        old = len(substring_match(paths, pf))
        raw = len(segment_match(paths, pf))
        new = len(segment_match(paths, pf if pf.endswith("/") else pf + "/"))
        print(f"{pf:<28}{old:>10}{raw:>12}{new:>12}{old - new:>7}")

from concurrent.futures import ThreadPoolExecutor
 
 
def fetch_level(store_id: str, prefix: str) -> list[dict]:
    """prefix 直下の 1 段だけを取る。ページングは全て消化して返す。"""
    items, cursor = [], None
    while True:
        page = client.beta.memory_stores.memories.list(
            memory_store_id=store_id, path_prefix=prefix,
            depth=1, view="basic", after_id=cursor, betas=BETAS,
        )
        items += [{"path": i.path, "type": i.type} for i in page.data]
        if not getattr(page, "has_more", False):
            return items
        cursor = page.last_id
 
 
def list_level_parallel(store_id: str, prefix: str = "/", workers: int = 8) -> list[str]:
    """木を段ごとに下る。プールの中で再帰しないため、ワーカー数に関わらず詰まらない。"""
    out: list[str] = []
    frontier = [normalize_prefix(prefix)]
    with ThreadPoolExecutor(max_workers=workers) as ex:
        while frontier:
            next_level: list[str] = []
            for page in ex.map(lambda p: fetch_level(store_id, p), frontier):
                for item in page:
                    if item["type"] == "memory_prefix":
                        next_level.append(item["path"])
                    else:
                        out.append(item["path"])
            frontier = next_level
    return sorted(out)

grep -rn 'path_prefix="[^"]*[^/]"' app

"""prefix_callsites.py — memories.list の path_prefix を ast で列挙し、
末尾スラッシュを補えていない呼び出しを洗い出す。"""
import ast, pathlib, sys
 
 
def prefix_arg(call):
    for kw in call.keywords:
        if kw.arg == "path_prefix":
            return kw.value
    return None
 
 
def literal_of(node):
    """定数なら文字列、f-string なら固定の末尾を返す。判定不能は None。"""
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value, "literal"
    if isinstance(node, ast.JoinedStr):
        tail = node.values[-1]
        if isinstance(tail, ast.Constant) and isinstance(tail.value, str):
            return tail.value, "fstring"
        return "", "fstring"          # 末尾が式 = スラッシュは付いていない
    return None, "dynamic"
 
 
def scan(path):
    tree = ast.parse(path.read_text(), filename=str(path))
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        if not (isinstance(node.func, ast.Attribute) and node.func.attr == "list"):
            continue
        arg = prefix_arg(node)
        if arg is None:
            continue
        value, kind = literal_of(arg)
        if kind == "dynamic":
            verdict = "REVIEW"        # 静的には決まらない。人が見る
        elif value.endswith("/"):
            verdict = "ok"
        else:
            verdict = "MISSING /"
        yield path.name, node.lineno, kind, verdict
 
 
if __name__ == "__main__":
    rows = [r for f in sorted(pathlib.Path(sys.argv[1]).rglob("*.py")) for r in scan(f)]
    print(f"{'file':<18}{'line':>5}  {'kind':<8}{'verdict'}")
    for f, ln, kind, verdict in rows:
        print(f"{f:<18}{ln:>5}  {kind:<8}{verdict}")
    bad = [r for r in rows if r[3] != "ok"]
    print(f"\ncall sites: {len(rows)}   needs attention: {len(bad)}")