{
  "task": "重构 user service 的鉴权逻辑",
  "decisions": [
    { "id": "d1", "text": "保留 /api/v1 接口签名", "reason": "下游 billing 服务依赖" }
  ],
  "files_touched": ["src/user/auth.ts", "src/middleware/jwt.ts"],
  "pending": ["补充 refresh token 的测试用例"],
  "constraints": ["不要引入新的运行时依赖"]
}

候选集 = 向量相似度 Top-K ∪ 关键词/BM25 匹配 Top-K
      ∪ 最近 N 条工作记忆
      ∪ 用户显式 pin 的条目
最终排序后取有限条注入上下文

.agent/
├── working.json          # 工作记忆（任务级，每轮覆盖）
├── memories.jsonl        # 长期记忆（追加写，一行一条）
└── config.json           # 配置（阈值、检索参数）

{
  "task": "",
  "decisions": [],
  "files_touched": [],
  "pending": [],
  "constraints": [],
  "updated_at": ""
}

import json
from pathlib import Path

WORKING = Path(".agent/working.json")

def load_working() -> dict:
    return json.loads(WORKING.read_text(encoding="utf-8"))

def render_working(w: dict) -> str:
    if not w.get("task"):
        return ""
    lines = [f"## 当前任务\n{w['task']}"]
    if w["decisions"]:
        lines.append("## 已定决策（不要推翻）")
        for d in w["decisions"]:
            lines.append(f"- {d['text']}（原因：{d['reason']}）")
    if w["constraints"]:
        lines.append("## 硬约束")
        lines += [f"- {c}" for c in w["constraints"]]
    if w["pending"]:
        lines.append("## 待办")
        lines += [f"- {p}" for p in w["pending"]]
    return "\n".join(lines)

TOOLS = [{
    "name": "update_working_memory",
    "description": "更新任务状态。做出关键决策、发现新约束、完成待办时调用。",
    "input_schema": {
        "type": "object",
        "properties": {
            "task": {"type": "string", "description": "任务目标，仅在任务变更时填写"},
            "add_decision": {
                "type": "object",
                "properties": {
                    "text": {"type": "string"},
                    "reason": {"type": "string", "description": "为什么这样选，必填"}
                },
                "required": ["text", "reason"]
            },
            "add_constraint": {"type": "string"},
            "resolve_pending": {"type": "string"},
            "add_pending": {"type": "string"},
            "touch_file": {"type": "string"}
        }
    }
}]

def handle_update(args: dict):
    w = load_working()
    if args.get("task"):
        w["task"] = args["task"]
    if d := args.get():
        w[].append({: , **d})
     c := args.get():
        w[].append(c)
     p := args.get():
        w[].append(p)
     p := args.get():
        w[] = [x  x  w[]  x != p]
     f := args.get():
         f   w[]:
            w[].append(f)
    w[] = now_iso()
    WORKING.write_text(json.dumps(w, ensure_ascii=, indent=))
     

当你要做出架构/接口/依赖层面的选择时，先调用 update_working_memory 记录决策和原因，
再继续执行。当用户纠正你时，同样记录一条决策，原因写「用户纠正」。

SUMMARY_PROMPT = """基于以下任务状态，生成一条记忆摘要。要求：
1. 3-5 句话
2. 必须包含：任务目标、最终方案、被否决的方案及原因
3. 不要包含过程性细节（执行了什么命令、试错了几次）
4. 输出 JSON：{"summary": "...", "decisions": [...], "tags": [...]}

任务状态：
{working}
"""

def save_memory(summary_obj: dict, base: dict):
    record = {
        "id": stable_hash(summary_obj["summary"]),
        "summary": summary_obj["summary"],
        "tags": summary_obj.get("tags", []),
        "files": base.get("files_touched", []),
        "created_at": now_iso(),
        "last_hit_at": None,     # 引用计数用
        "hit_count": 0,
        "status": "active"       # active | deprecated
    }
    with open(".agent/memories.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

def recall(task_desc: str, top_k: int = 3) -> list[dict]:
    records = [json.loads(l) for l in
               Path(".agent/memories.jsonl").read_text(encoding="utf-8").splitlines()
               if l.strip()]
    records = [r for r in records if r["status"] == "active"]

    vec_hits = vector_search(task_desc, records, k=top_k)
    kw_hits  = bm25_search(task_desc, records, k=top_k)
    pinned   = [r for r in records if r.get("pinned")]

    merged, seen = [], set()
    for r in vec_hits + kw_hits + pinned:
        if r["id"] not in seen:
            merged.append(r); seen.add(r["id"])

    # 时效与引用加权
    merged.sort(key=lambda r: score(r, task_desc), reverse=True)
    result = merged[:top_k]

    bump_hits(result)   # 命中就更新 hit_count / last_hit_at
    return result

def bump_hits(records):
    ids = {r["id"] for r in records}
    lines = Path(".agent/memories.jsonl").read_text(encoding="utf-8").splitlines()
    out = []
    for l in lines:
        if not l.strip():
            continue
        r = json.loads(l)
        if r["id"] in ids:
            r["hit_count"] += 1
            r["last_hit_at"] = now_iso()
        out.append(json.dumps(r, ensure_ascii=False))
    Path(".agent/memories.jsonl").write_text("\n".join(out) + "\n", encoding="utf-8")

def render_recall(records: list[dict]) -> str:
    if not records:
        return ""
    lines = ["## 相关历史经验（供参考，可能已过时）"]
    for r in records:
        lines.append(f"- {r['summary']}  _({r['created_at'][:10]})_")
    return "\n".join(lines)

def decay_and_gc():
    records = load_all()
    now = datetime.now(timezone.utc)
    for r in records:
        age_days = (now - parse(r["created_at"])).days
        idle_days = (now - parse(r["last_hit_at"] or r["created_at"])).days

        # 长期零引用 → 归档
        if r["hit_count"] == 0 and age_days > 90:
            r["status"] = "archived"

        # 长期未被命中 → 降权（在检索打分时体现）
        r["weight"] = 1.0 if idle_days < 14 else 0.5

    # 被推翻的记忆由人工或 Agent 显式标记
    save_all(records)

def score(r, task_desc):
    base = similarity(r, task_desc)
    freshness = 1.0 / (1 + days_since(r["created_at"]) / 30)
    return base * r.get("weight", 1.0) * freshness

log.info("injected_memory ids=%s source=%s tokens=%d",
         [r["id"][:8] for r in recall_result],
         "recall" if recall_result else "none",
         est_tokens(render_recall(recall_result)))