harness/phase4_overlap.py

"""Phase 4.3 — shared-user-task personalization overlap sweep.

Stress-tests the personalization claim (per-user memory vs. a live global/shared
store) as users' task sets *overlap*. Workload: every user has ``tasks_per_user``
signature tasks with a FIXED per-user argument realization (user A always NYC,
user B always Boston), consistent across the user's sessions. A fraction
``overlap_frac`` of those tasks come from ONE shared template pool that every
user reuses (same templates, different per-user args); the rest are private and
disjoint. As overlap rises, a global store sees many users' different calls for
the same template — does per-user memory still win, or does a shared store catch
up?

Two arms, both live + evicting (only partitioning differs), so the contrast is
personalization alone:
  - global_evict  : one live global store (no per-user view)
  - personal_memory (ours): per-user live evicting store

Targets are the served model's greedy calls (real, generated once per unique
query on the migrated node and cached per overlap level). Run from ``code/``:
    python -m harness.phase4_overlap --url http://localhost:30000/v1
"""
from __future__ import annotations

import argparse
import json
import os
from collections import defaultdict
from pathlib import Path

from . import metrics
from .client import ToolClient
from .data import load_bfcl
from .memory import Embedder, GlobalEvict, PersonalMemory
from .run_accept import _parse_target, generate_targets
from .simulate import build_users

ROOT = Path(__file__).resolve().parent.parent
RESULTS = ROOT / "results"
# Tokenizer for the token-LCP accept metric: HF hub id by default;
# override with a local snapshot path if running offline.
MODEL_PATH = os.environ.get("SPECMEM_TOKENIZER", "openai/gpt-oss-120b")


class Slot:
    """generate_targets expects objects with .query and .functions; the
    functions registry is identical across BFCL tasks' first-tool schema, so we
    attach it lazily below."""
    __slots__ = ("query", "functions")

    def __init__(self, query, functions=None):
        self.query = query
        self.functions = functions


def _replay(instances, targets, embedder, capacity):
    arms = [GlobalEvict(capacity=40 * capacity),
            PersonalMemory(capacity=capacity, eviction="lru")]
    agg = {a.name: defaultdict(list) for a in arms}
    for ins in instances:
        tgt = targets.get(ins.query)
        if tgt is None:
            continue
        for a in arms:
            agg[a.name][ins.session].append(metrics.score(
                a.draft(ins.query, ins.functions, ins.user_id, embedder), tgt))
        cn, ca = _parse_target(tgt)
        for a in arms:
            a.observe(ins.query, ins.functions, ins.user_id, cn, ca, embedder)
    out = {}
    for n, v in agg.items():
        post = [x for s, xs in v.items() if s > 0 for x in xs]
        out[n] = round(sum(x["accept_length"] for x in post) /
                       max(1, len(post)), 3)
    return out


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--url", default="http://localhost:30000/v1")
    p.add_argument("--model", default="gpt-oss-120b")
    p.add_argument("--workers", type=int, default=16)
    p.add_argument("--overlaps", nargs="+", type=float,
                   default=[0.0, 0.25, 0.5, 0.75, 1.0])
    p.add_argument("--seeds", nargs="+", type=int, default=[0, 1, 2])
    p.add_argument("--capacity", type=int, default=48)
    args = p.parse_args()

    metrics.get_tokenizer(MODEL_PATH)
    tasks = load_bfcl()
    embedder = Embedder()
    client = ToolClient(url=args.url, model=args.model)
    if not client.ping():
        raise SystemExit(f"served model not reachable at {args.url}")

    sweep = {}
    for ov in args.overlaps:
        per_seed = {"global_evict": [], "personal_memory": []}
        for sd in args.seeds:
            instances = build_users(
                tasks, n_users=40, tasks_per_user=15, n_sessions=12,
                queries_per_session=6, seed=sd, overlap_frac=ov,
                user_consistent=True)
            # attach the (shared) functions registry to Slots for target gen
            fmap = {ins.query: ins.functions for ins in instances}
            cache_f = RESULTS / f"phase4_overlap_targets_ov{ov}_seed{sd}.json"
            cache = json.loads(cache_f.read_text()) if cache_f.exists() else {}
            miss = [Slot(q, fmap[q]) for q in {i.query for i in instances}
                    if q not in cache]
            if miss:
                cache.update(generate_targets(client, miss, workers=args.workers))
                cache_f.write_text(json.dumps(cache))
            res = _replay(instances, cache, embedder, args.capacity)
            per_seed["global_evict"].append(res["global_evict"])
            per_seed["personal_memory"].append(res["personal_memory"])
            print(f"ov={ov} seed={sd}: {res}", flush=True)
        import statistics as st
        g = st.mean(per_seed["global_evict"])
        pm = st.mean(per_seed["personal_memory"])
        sweep[ov] = {
            "global_evict": round(g, 3),
            "personal_memory": round(pm, 3),
            "personal_advantage_pct": round(100 * (pm - g) / g, 1),
            "per_seed": per_seed}
        print(f"== ov={ov}: global {g:.3f} | personal {pm:.3f} | "
              f"adv {sweep[ov]['personal_advantage_pct']}% ==", flush=True)

    out = {"config": {"users": 40, "tasks_per_user": 15, "sessions": 12,
                      "capacity": args.capacity, "seeds": args.seeds,
                      "user_consistent_args": True,
                      "arms": "global_evict (live shared) vs personal_memory (ours)"},
           "sweep": sweep}
    (RESULTS / "phase4_personalization_overlap.json").write_text(
        json.dumps(out, indent=2))
    print(json.dumps({ov: {"global": s["global_evict"],
                           "personal": s["personal_memory"],
                           "adv%": s["personal_advantage_pct"]}
                      for ov, s in sweep.items()}, indent=1))


if __name__ == "__main__":
    main()