記憶が消えたのではなく、別のプロファイルに書かれていました — Memory profiles のスコープキーを実測で正す

def naive(s):
    return json.dumps(s)
 
def canon(s):
    # A: よく勧められる形。キー順とスペースを揃える
    return json.dumps(s, sort_keys=True, separators=(",", ":"))
 
def canon_strip(s):
    # B: A + 値の前後空白を落とす
    return json.dumps(
        {k: (v.strip() if isinstance(v, str) else v) for k, v in s.items()},
        sort_keys=True, separators=(",", ":"),
    )
 
def canon_type(s):
    # C: B + 型を文字列へ寄せる
    return json.dumps(
        {k: str(v).strip() for k, v in s.items()},
        sort_keys=True, separators=(",", ":"),
    )
 
ALLOW = ("app_id", "user_id")
 
def canon_allow(s):
    # D: C + スコープに入れてよいキーを固定する
    return json.dumps(
        {k: str(s[k]).strip() for k in ALLOW if k in s},
        sort_keys=True, separators=(",", ":"),
    )

def run_with_fallback(calls, norm):
    store = {}
    hits = miss = leaks = 0
    for app, u, kind, s, tier in calls:
        k = (SCHEMA, norm(s))
        p = store.get(k)
        if p is None:
            # 見つからなければ user_id が一致する既存プロファイルを探す
            want = str(s.get("user_id", "")).strip()
            for (_sch, kk), v in store.items():
                try:
                    kd = json.loads(kk)
                except Exception:
                    continue
                if str(kd.get("user_id", "")).strip() == want:
                    p = v
                    if v["owner"] != (app, u):
                        leaks += 1        # 別テナントの記録を返した
                    break
        if p is None:
            miss += 1
            store[k] = {"owner": (app, u), "billing_plan": tier}
        else:
            hits += 1
    n = hits + miss
    return {"hit": 100 * hits / n, "profiles": len(store), "leaks": leaks}

到達しうる参照率 = 1 - (固有スコープ数 / ターン数)

# 最初に書いた版 — 動くが、探している異常を見逃す
ceiling = 100 * (1 - len(store) / turns)
if observed_hit > ceiling:
    alert("scope が併合されている可能性")

def health(observed_hit_pct, turns, distinct_tenants):
    """天井は store の状態ではなく、独立に数えたテナント数から出す。
    store 由来で計算すると、併合が起きたときに天井も一緒に下がり
    アラームが沈黙する（実測で確認済み）。"""
    ceiling = 100 * (1 - distinct_tenants / turns)
    return {
        "ceiling": ceiling,
        "merged": observed_hit_pct > ceiling + 1e-9,       # 境界が消えている
        "fragmented": observed_hit_pct < ceiling - 5.0,    # スコープが分裂している
    }

#!/usr/bin/env python3
"""scope drift lint — 同じ scope キーの書かれ方が場所によってぶれている箇所を指摘する。
絶対的な正しさは判定しない。多数派と異なる書き方だけを不一致として報告する。"""
import ast
import sys
import pathlib
from collections import defaultdict
 
ALLOWED = {"app_id", "user_id"}
SINKS = {"ingest", "retrieve_profile", "generate_memories"}
 
 
def shape(v):
    """値の「書かれ方」を分類する。str() 済みか、素の変数か、定数か。"""
    if isinstance(v, ast.Call) and isinstance(v.func, ast.Name) and v.func.id == "str":
        return "str()"
    if isinstance(v, ast.Name):
        return "bare-name"
    if isinstance(v, ast.JoinedStr):
        return "f-string"
    if isinstance(v, ast.Constant):
        return f"const:{type(v.value).__name__}"
    return "other"
 
 
class Collect(ast.NodeVisitor):
    """scope= に渡される辞書リテラルを集める。変数経由の1段も追う。"""
 
    def __init__(self):
        self.sites = []
        self.assigned = {}
 
    def visit_Assign(self, n):
        if (isinstance(n.value, ast.Dict) and len(n.targets) == 1
                and isinstance(n.targets[0], ast.Name)):
            self.assigned[n.targets[0].id] = n.value
        self.generic_visit(n)
 
    def visit_Call(self, n):
        fname = n.func.attr if isinstance(n.func, ast.Attribute) else getattr(n.func, "id", None)
        if fname in SINKS:
            for kw in n.keywords:
                if kw.arg != "scope":
                    continue
                d = kw.value if isinstance(kw.value, ast.Dict) \
                    else self.assigned.get(getattr(kw.value, "id", None))
                if isinstance(d, ast.Dict):
                    self.sites.append(d)
        self.generic_visit(n)
 
 
def scan(root):
    sites = []
    for p in sorted(pathlib.Path(root).rglob("*.py")):
        try:
            tree = ast.parse(p.read_text(encoding="utf-8"))
        except SyntaxError:
            continue
        c = Collect()
        c.visit(tree)
        sites += [(str(p), d) for d in c.sites]
 
    # 第1パス: キーごとに「どの書かれ方が何回出たか」を数え、多数派を決める
    shapes = defaultdict(lambda: defaultdict(int))
    for _, d in sites:
        for k, v in zip(d.keys, d.values):
            if isinstance(k, ast.Constant) and k.value in ALLOWED:
                shapes[k.value][shape(v)] += 1
    majority = {k: max(s.items(), key=lambda x: x[1])[0] for k, s in shapes.items()}
 
    # 第2パス: 多数派から外れた箇所と、許可外キー・必須キー欠落を報告する
    out = []
    for path, d in sites:
        keys = []
        for k, v in zip(d.keys, d.values):
            if not (isinstance(k, ast.Constant) and isinstance(k.value, str)):
                out.append((path, d.lineno, "dynamic-key", "キーが定数文字列でない"))
                continue
            keys.append(k.value)
            if k.value not in ALLOWED:
                out.append((path, d.lineno, "extra-key",
                            f"許可外のキー {k.value!r} — プロファイルが分裂します"))
                continue
            sh = shape(v)
            if sh != majority[k.value]:
                out.append((path, d.lineno, "shape-drift",
                            f"{k.value} が {sh}（コーパス多数派は {majority[k.value]}）"))
            if sh.startswith("const:") and sh != "const:str":
                out.append((path, d.lineno, "type-drift", f"{k.value} が文字列でない"))
        for m in sorted(ALLOWED - set(keys)):
            out.append((path, d.lineno, "missing-key", f"必須キー欠落 {m!r}"))
    return out
 
 
if __name__ == "__main__":
    rows = scan(sys.argv[1] if len(sys.argv) > 1 else ".")
    for f, l, kind, msg in rows:
        print(f"{f}:{l}: [{kind}] {msg}")
    print(f"-- {len(rows)} 件", file=sys.stderr)
    sys.exit(1 if rows else 0)   # CI で落とす

repo/a.py:4: [shape-drift] user_id が bare-name（コーパス多数派は str()）
repo/b.py:2: [extra-key] 許可外のキー 'device_id' — プロファイルが分裂します
repo/c.py:2: [shape-drift] user_id が f-string（コーパス多数派は str()）
repo/c.py:3: [extra-key] 許可外のキー 'tier' — プロファイルが分裂します
repo/c.py:4: [extra-key] 許可外のキー 'locale' — プロファイルが分裂します
-- 5 件

from dataclasses import dataclass
import json
 
@dataclass(frozen=True)
class Tenant:
    """スコープになりうる値を型として持つ。文字列を直接受け取らない。"""
    app_id: str
    user_id: str
 
    def __post_init__(self):
        for name in ("app_id", "user_id"):
            v = getattr(self, name)
            if not isinstance(v, str) or v != v.strip() or not v:
                raise ValueError(f"{name} は前後空白のない非空の文字列である必要があります: {v!r}")
 
 
def scope_of(t: Tenant) -> dict:
    """スコープを組み立てる唯一の入口。ここ以外で辞書を作らない。"""
    return {"app_id": t.app_id, "user_id": t.user_id}
 
 
def scope_key(t: Tenant) -> str:
    """保存先の同一性を決めるキー。ログや監視でも同じ関数を使う。"""
    return json.dumps(scope_of(t), sort_keys=True, separators=(",", ":"))