diff --git a/README.md b/README.md
index cbba20d..c936625 100644
--- a/README.md
+++ b/README.md
@@ -52,6 +52,7 @@ https://github.com/user-attachments/assets/49c1a306-d2e9-49d6-9c83-65e38a62df30
 
 ## 📣 News
 
+- **2026-06-30** — **Arbor learns from its own runs.** Each run leaves concrete, reusable findings — a dataset quirk that helped, a trap to avoid; the next similar task recalls them at intake, so the agent starts from experience instead of scratch. 🧠
 - **2026-06-22** — **Built-in literature search & idea novelty checks.** Arbor can now ground its research in prior work via the public [alphaXiv](https://www.alphaxiv.org) API — zero config, no search endpoint or key. Novelty-check any idea before you build it with `arbor idea-check "<your idea>"`, or let the Coordinator vet every new branch automatically. See [Literature Search & Novelty Checks](#-literature-search--novelty-checks). 🔎
 - **2026-06-18** — Arbor was featured by [VentureBeat](https://venturebeat.com/), one of the leading tech media outlets in the US: ["New AI optimization framework beats Claude Code and Codex by 2.5x on the same compute budget"](https://venturebeat.com/orchestration/new-ai-optimization-framework-beats-claude-code-and-codex-by-2-5x-on-the-same-compute-budget). 📰
 - **2026-06-12** — Arbor's native CLI runtime and Agent Skill Suite (Codex / Claude Code) are released. 🚀
diff --git a/README.zh-CN.md b/README.zh-CN.md
index c9e15f4..c166fd6 100644
--- a/README.zh-CN.md
+++ b/README.zh-CN.md
@@ -46,6 +46,7 @@
 
 ## 📣 最新动态
 
+- **2026-06-30** — **Arbor 会从自己的运行中学习。** 每次运行都会留下具体、可复用的发现——数据集上能提分的特性、需要避开的坑；下次类似任务在 intake 时把它们召回，于是 agent 从经验出发，而不是从零开始。🧠
 - **2026-06-22** — **内置文献检索与想法新颖性审查。** Arbor 现在可以通过 [alphaXiv](https://www.alphaxiv.org) 公共 API 把研究建立在已有工作之上——零配置，无需搜索端点或密钥。动手前用 `arbor idea-check "<你的想法>"` 审查任意想法的新颖性，或让 Coordinator 自动为每个新分支把关。详见[文献检索与新颖性审查](#-文献检索与新颖性审查)。🔎
 - **2026-06-18** — Arbor 被美国知名科技媒体 [VentureBeat](https://venturebeat.com/) 报道：[《New AI optimization framework beats Claude Code and Codex by 2.5x on the same compute budget》](https://venturebeat.com/orchestration/new-ai-optimization-framework-beats-claude-code-and-codex-by-2-5x-on-the-same-compute-budget)。📰
 - **2026-06-12** — Arbor 原生 CLI 运行时与智能体技能套件（Codex / Claude Code）正式发布。🚀
diff --git a/src/cli/intake/launch_tool.py b/src/cli/intake/launch_tool.py
index de83e84..495cc5a 100644
--- a/src/cli/intake/launch_tool.py
+++ b/src/cli/intake/launch_tool.py
@@ -133,6 +133,17 @@ class LaunchExperimentTool(Tool):
                 "items": {"type": "string"},
                 "description": "Optional bullet notes (constraints, edge cases) to attach to the plan.",
             },
+            "apply_experience": {
+                "type": "boolean",
+                "description": "Set true when the user agreed to reuse prior experience; "
+                               "matched lessons are composed into the instruction.",
+            },
+            "experience_sessions": {
+                "type": "array",
+                "items": {"type": "string"},
+                "description": "Session names (from 'Prior experience available') you judged "
+                               "relevant to this goal. You select; keyword match is only a fallback.",
+            },
         },
         "required": ["cwd", "instruction"],
     }
@@ -169,7 +180,9 @@ async def execute(self, **kwargs: Any) -> str:
 
         plan = LaunchPlan(
             cwd=str(resolved.resolve()),
-            instruction=instruction,
+            instruction=_with_experience(str(resolved.resolve()), instruction,
+                                         kwargs.get("apply_experience"),
+                                         kwargs.get("experience_sessions")),
             rationale=(kwargs.get("rationale") or "").strip(),
             suggested_max_cycles=_safe_int(kwargs.get("suggested_max_cycles")),
             suggested_max_turns=_safe_int(kwargs.get("suggested_max_turns")),
@@ -192,3 +205,23 @@ def _safe_int(v: Any) -> int | None:
         return int(v) if v is not None else None
     except (TypeError, ValueError):
         return None
+
+
+def _with_experience(cwd: str, instruction: str, apply: Any, sessions: Any = None) -> str:
+    """Prepend composed prior experience to the instruction when the user opted in.
+
+    Prefers the sessions the intake agent (an LLM) selected as relevant; falls back
+    to keyword topic matching only if it named none.
+    """
+    if not apply:
+        return instruction
+    try:
+        from ...recall import compose_for_topic, compose_from_sessions
+        block = ""
+        if isinstance(sessions, list) and sessions:
+            block = compose_from_sessions(cwd, [str(s) for s in sessions])
+        if not block:
+            block = compose_for_topic(cwd, instruction)
+    except Exception:  # pylint: disable=broad-exception-caught
+        block = ""
+    return f"{block}\n\n---\n{instruction}" if block else instruction
diff --git a/src/cli/intake/system_prompt.py b/src/cli/intake/system_prompt.py
index 30b2ce5..f0d3325 100644
--- a/src/cli/intake/system_prompt.py
+++ b/src/cli/intake/system_prompt.py
@@ -3,6 +3,28 @@
 from __future__ import annotations
 
 
+def _prior_experience_block(starting_cwd: str) -> str:
+    """If past runs here left experience, tell intake to offer it and ask the user."""
+    try:
+        from ...recall import list_experiences
+        exps = list_experiences(starting_cwd)
+    except Exception:  # pylint: disable=broad-exception-caught
+        exps = []
+    if not exps:
+        return ""
+    lines = [f"  - {name}: {desc}" for name, desc in exps]
+    return (
+        "## Prior experience available\n"
+        "This project has experience from past runs:\n"
+        + "\n".join(lines) + "\n"
+        "If the user's goal is similar, briefly offer to reuse the relevant one "
+        "and ask whether to apply it (default yes). Don't push it if unrelated.\n"
+        "When launching with the user's agreement, set apply_experience=true and "
+        "list the relevant session names in experience_sessions (you judge relevance).\n"
+        "\n"
+    )
+
+
 def build_system_prompt(*, starting_cwd: str) -> str:
     """Build the planning-agent system prompt.
 
@@ -101,6 +123,7 @@ def build_system_prompt(*, starting_cwd: str) -> str:
         "## Target directory\n"
         f"The user launched you from: {starting_cwd}\n"
         "\n"
+        + _prior_experience_block(starting_cwd) +
         "Treat that path as the default target. In the common case the user "
         "is already inside their project — just go with it and start talking "
         "about the goal. Quietly check it looks like a project (one Glob is "
diff --git a/src/coordinator/config.py b/src/coordinator/config.py
index c1fc22d..436a67b 100644
--- a/src/coordinator/config.py
+++ b/src/coordinator/config.py
@@ -476,6 +476,10 @@ class CoordinatorConfig(ProxyModel):
     export_trajectory: bool = True
     # Capture per-call token-level traces (tokens.jsonl) for SFT/RL. Heavy; off.
     token_trace: bool = False
+    # Distill run insights into a reusable skill in ~/.arbor/skills (line 2). Off.
+    distill_skills: bool = False
+    # Use the LLM to abstract distilled lessons into transferable principles. Off.
+    distill_abstract: bool = False
 
     # ── Plugin (runtime object; not serialized) ──────────────────────
     plugin: Any = PydField(default=None, exclude=True, repr=False)
diff --git a/src/coordinator/orchestrator.py b/src/coordinator/orchestrator.py
index 0e47377..6d43290 100644
--- a/src/coordinator/orchestrator.py
+++ b/src/coordinator/orchestrator.py
@@ -1154,6 +1154,16 @@ def _write_run_stats(
             except Exception as e:  # pylint: disable=broad-exception-caught
                 _print_status(f"Warning: failed to write trajectory: {e}")
 
+        # Self-evolution line 2: distill insights into a reusable cross-run skill.
+        if getattr(self.config, "distill_skills", False):
+            try:
+                from ..distill import distill_to_session
+                prov = self.provider if getattr(self.config, "distill_abstract", False) else None
+                p = distill_to_session(self.config.workspace_dir, provider=prov)
+                _print_status(f"Distilled experience: {p}" if p else "Distill: nothing to learn")
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                _print_status(f"Warning: failed to distill skill: {e}")
+
     # ------------------------------------------------------------------
     # Final report
     # ------------------------------------------------------------------
diff --git a/src/coordinator/tools/__init__.py b/src/coordinator/tools/__init__.py
index d1cf80f..434cf36 100644
--- a/src/coordinator/tools/__init__.py
+++ b/src/coordinator/tools/__init__.py
@@ -19,6 +19,7 @@
 from .search_ctx import SearchIdeaContextTool, SearchIdeaContextParallelTool, SearchStatusTool
 from .research_ctx import ResearchSearchTool
 from .ask_user import AskUserTool
+from .record_finding import RecordFindingTool
 from ..convergence import ConvergenceDetector, ConvergenceConfig
 
 if TYPE_CHECKING:
@@ -85,6 +86,7 @@ def get_coordinator_tools(
         FileReadTool(cwd=cwd, workspace_dir=wdir),
         GrepTool(cwd=cwd, workspace_dir=wdir),
         GlobTool(cwd=cwd, workspace_dir=wdir),
+        RecordFindingTool(cwd=cwd, workspace_dir=wdir),
     ]
     if skill_registry is not None:
         # Skills (read-only reference docs loaded on demand)
diff --git a/src/coordinator/tools/record_finding.py b/src/coordinator/tools/record_finding.py
new file mode 100644
index 0000000..f239033
--- /dev/null
+++ b/src/coordinator/tools/record_finding.py
@@ -0,0 +1,55 @@
+"""RecordFinding — let the agent jot a concrete, situational discovery mid-run.
+
+This is the live half (A) of experience capture. When the coordinator or executor
+notices something specific and worth remembering for next time — a dataset quirk
+that helps the metric, a trap an executor fell into, a gotcha in the harness — it
+records it here. Kept specific on purpose; finalize also mines the trajectory (B)
+for findings that were never explicitly logged.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ...core.tools.base import Tool
+
+
+class RecordFindingTool(Tool):
+    """Record a concrete discovery or pitfall for future runs to reuse."""
+
+    name = "RecordFinding"
+    is_read_only = True  # only appends to a notes file; safe to parallelize
+    description = (
+        "Record a CONCRETE, situational discovery worth remembering next time you "
+        "work on this dataset / task / harness. Two typical kinds:\n"
+        "- 'leverage': a specific property you found that helps the metric "
+        "(e.g. 'this dataset's labels are noisy above index 9000 — drop them').\n"
+        "- 'pitfall': a specific trap to avoid (e.g. 'the executor kept editing "
+        "eval.py — remind it the harness is protected').\n"
+        "Keep it specific — the value is the detail, not a general principle. Do "
+        "NOT log routine progress or generic advice."
+    )
+    input_schema: dict[str, Any] = {
+        "type": "object",
+        "properties": {
+            "kind": {"type": "string", "description": "'leverage' or 'pitfall' (free-form ok)."},
+            "about": {"type": "string", "description": "What it concerns: the dataset, an executor, the harness, etc."},
+            "note": {"type": "string", "description": "The concrete finding, in one or two sentences."},
+        },
+        "required": ["note"],
+    }
+
+    def __init__(self, *, cwd: str, workspace_dir: str | None = None):
+        super().__init__(cwd=cwd, workspace_dir=workspace_dir)
+
+    async def execute(self, **kwargs: Any) -> str:
+        note = (kwargs.get("note") or "").strip()
+        if not note:
+            return "Error: 'note' is required."
+        try:
+            from ...experience import record_finding
+            record_finding(self.workspace_dir or self.cwd,
+                           kind=kwargs.get("kind", ""), about=kwargs.get("about", ""), note=note)
+        except Exception as exc:  # pylint: disable=broad-exception-caught
+            return f"Could not record finding: {exc}"
+        return f"Recorded {kwargs.get('kind') or 'finding'}: {note[:80]}"
diff --git a/src/coordinator/tools/tree_ops.py b/src/coordinator/tools/tree_ops.py
index 27e2c4d..e1bfb43 100644
--- a/src/coordinator/tools/tree_ops.py
+++ b/src/coordinator/tools/tree_ops.py
@@ -331,6 +331,22 @@ async def execute(self, **kwargs: Any) -> str:
 
         self._tree.update_node(node_id, **updates)
 
+        # Live experience capture (self-evolution): log lessons as the research
+        # advances, not only at finalize. Best-effort — never break a tool call.
+        if updates.get("insight") or updates.get("status"):
+            try:
+                from ...experience import append_experience
+                # Use the node's current insight (it may have been set by a prior
+                # update), not just this call's fields, so notes aren't empty.
+                node = self._tree.get_node(node_id)
+                enriched = dict(updates)
+                if not enriched.get("insight") and node is not None and getattr(node, "insight", ""):
+                    enriched["insight"] = node.insight
+                append_experience(getattr(self._tree, "json_path", None),
+                                   node_id=node_id, updates=enriched)
+            except Exception:  # pylint: disable=broad-exception-caught
+                pass
+
         parts = [f"Updated {node_id}:"]
         for k, v in updates.items():
             parts.append(f"  {k} = {v}")
diff --git a/src/distill.py b/src/distill.py
new file mode 100644
index 0000000..6dba38d
--- /dev/null
+++ b/src/distill.py
@@ -0,0 +1,172 @@
+"""Experience distillation (self-evolution line 2).
+
+Turns a finished run into a consolidated EXPERIENCE.md in its own session folder,
+layered by altitude (meta = cross-domain strategy; domain = idea-class wins/losses).
+Task-specific findings stay in REPORT/trajectory. Experience is recalled per-session
+by recall.find_similar, not registered as global skills. Best-effort: never fail a run.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any
+
+from .trajectory import _load_tree
+
+
+def _slug(s: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "-", (s or "general").lower()).strip("-") or "general"
+
+
+def _domain(tree_meta: dict[str, Any], session_dir: Path) -> str:
+    d = tree_meta.get("domain") or tree_meta.get("benchmark") or tree_meta.get("cwd")
+    if not d:
+        # session lives at <project>/.arbor/sessions/<run>; recover the project name.
+        parts = session_dir.resolve().parts
+        d = parts[-4] if len(parts) >= 4 and parts[-3:-1] == ("sessions",) else ""
+        if not d and ".arbor" in parts:
+            d = parts[parts.index(".arbor") - 1]
+    return _slug(Path(str(d)).name) if d else "general"
+
+
+def _frag(name: str, desc: str, when: str, title: str, body: list[str]) -> str:
+    return "\n".join(["---", f"name: {name}", f"description: {desc}",
+                      f"when_to_apply: {when}", "---", f"\n# {title}\n", *body])
+
+
+def build_skills(session_dir: Path) -> list[tuple[str, str, str]]:
+    """Return [(level, domain, markdown), ...] — empty if nothing worth keeping."""
+    session_dir = Path(session_dir)
+    tree = _load_tree(session_dir)
+    if not tree:
+        return []
+    root = tree.get("ROOT") or {}
+    meta = root.get("meta", {}) if isinstance(root.get("meta"), dict) else {}
+    domain = _domain(meta, session_dir)
+    out: list[tuple[str, str, list[str]]] = []  # (level, domain, bullets)
+
+    # domain layer — verified wins/losses, transferable within domain
+    wins = [n for n in tree.values() if n.get("status") in ("merged", "done") and n.get("insight")]
+    dom_bul = []
+    if root.get("insight"):
+        dom_bul.append(root["insight"].strip().split("\n")[0][:200])
+    for n in sorted(wins, key=lambda x: x.get("score") or 0, reverse=True):
+        dom_bul.append(f"[{n.get('status')}, score={n.get('score')}] {n.get('insight','').strip()[:200]}")
+    if dom_bul:
+        out.append(("domain", domain, dom_bul))
+
+    # meta layer — strategy from the tree's shape, transfers across domains
+    pruned = [n for n in tree.values() if n.get("status") == "pruned"]
+    process = [f"dead-end: {n.get('insight','').strip()[:160]}" for n in pruned if n.get("insight")]
+    try:
+        from .experience import load_experience
+        trail = [e for e in load_experience(session_dir) if e.get("status") in ("pruned", "done")]
+        process += [f"{e['node_id']}: {e['insight'][:140]}" for e in trail if e.get("insight")]
+    except Exception:  # pylint: disable=broad-exception-caught
+        pass
+    if process:
+        out.append(("meta", "general", process))
+    return out
+
+
+def _run_coro(coro: Any) -> Any:
+    """Run an async provider call from sync finalize, even inside a live loop.
+
+    ``asyncio.run`` raises if a loop is already running (which is why the earlier
+    abstraction pass silently fell back). Run it in a dedicated thread instead.
+    """
+    import asyncio
+    import concurrent.futures
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(coro)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
+        return ex.submit(lambda: asyncio.run(coro)).result()
+
+
+def _raw_lessons(frags: list[tuple[str, str, list[str]]]) -> list[str]:
+    """Flatten layered bullets into raw lessons, dropping markdown-heading noise."""
+    raw: list[str] = []
+    for _level, _d, bullets in frags:
+        for b in bullets:
+            b = b.strip()
+            if b and not b.lstrip("[").startswith("#") and "##" not in b:
+                raw.append(b)
+    return raw
+
+
+_MINE_SYS = (
+    "You review one optimization run and surface CONCRETE, situational findings worth "
+    "remembering for the next run on this same dataset / task / harness. Keep them "
+    "SPECIFIC — a dataset quirk that helped the metric, a trap an executor or the harness "
+    "fell into. NOT generic advice or principles. One finding per line, format:\n"
+    "[leverage|pitfall] SUBJECT: the concrete finding\n"
+    "where SUBJECT is the specific thing it concerns (the dataset, an executor, the "
+    "harness, a numpy call...). Return only such lines, or nothing if none."
+)
+
+
+def _mine_findings(provider: Any, raw: list[str]) -> list[dict[str, str]]:
+    """B: mine the run's lessons for concrete findings not explicitly logged."""
+    if not provider or not raw:
+        return []
+    try:
+        msg = "Run material:\n" + "\n".join(f"- {b}" for b in raw)
+        resp = _run_coro(provider.create(system=_MINE_SYS,
+                                         messages=[{"role": "user", "content": msg}], max_tokens=700))
+        found = []
+        for ln in resp.get_text().splitlines():
+            m = re.match(r"\s*\[?(leverage|pitfall)\]?\s*([^:]*):\s*(.+)", ln, re.I)
+            if m:
+                about = m.group(2).strip()
+                if about.lower() in ("about", "subject"):  # guard leaked placeholder
+                    about = ""
+                found.append({"kind": m.group(1).lower(), "about": about, "note": m.group(3).strip()})
+        return found
+    except Exception:  # pylint: disable=broad-exception-caught
+        return []
+
+
+def distill_to_session(session_dir: Path, provider: Any = None) -> Path | None:
+    """Write EXPERIENCE.md: the run's concrete findings (logged live + mined).
+
+    Experience here is specific and situational by design — dataset quirks that
+    help, traps to avoid — for the next run on the same/similar target. Combines
+    findings the agent logged via RecordFinding (A) with an LLM mining pass (B).
+    """
+    session_dir = Path(session_dir)
+    domain = _domain({}, session_dir)
+
+    findings: list[dict[str, str]] = []
+    seen: set[str] = set()
+
+    def _add(f: dict[str, str]) -> None:
+        note = (f.get("note") or "").strip()
+        key = re.sub(r"\W+", "", note.lower())[:60]
+        if note and key not in seen:
+            seen.add(key)
+            findings.append({"kind": f.get("kind", ""), "about": f.get("about", ""), "note": note})
+
+    try:  # A: explicitly logged findings
+        from .experience import load_findings
+        for f in load_findings(session_dir):
+            _add(f)
+    except Exception:  # pylint: disable=broad-exception-caught
+        pass
+
+    raw = _raw_lessons(build_skills(session_dir))  # B: mine the rest from the run
+    for f in _mine_findings(provider, raw):
+        _add(f)
+
+    if not findings:
+        return None
+    lines = [f"- **[{(f['kind'] or 'finding')}] {f['about']}** — {f['note']}" if f["about"]
+             else f"- **[{f['kind'] or 'finding'}]** {f['note']}" for f in findings]
+    md = _frag(f"experience-{domain}", f"Concrete findings from a {domain} run.",
+               "reuse when working on this dataset / task / harness again",
+               f"Findings: {domain}", lines)
+    out = session_dir / "EXPERIENCE.md"
+    out.write_text(md, encoding="utf-8")
+    return out
diff --git a/src/experience.py b/src/experience.py
new file mode 100644
index 0000000..737aa74
--- /dev/null
+++ b/src/experience.py
@@ -0,0 +1,89 @@
+"""Experience capture (self-evolution): two halves.
+
+1. **Live** — ``append_experience`` logs a lesson each time the coordinator
+   updates a node mid-run (status/insight), so the research *process* is kept,
+   not just the surviving tree.
+2. **Consolidate** — ``load_experience`` reads the notes back at finalize; the
+   distiller folds them together with the hypothesis tree into reusable skills.
+
+The session dir is recovered from the tree's json_path (<session>/.coordinator/
+idea_tree.json). Best-effort throughout — never break a run.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+EXPERIENCE_FILENAME = "experience.jsonl"
+
+
+def _session_dir(tree_json_path: str | None) -> Path | None:
+    if not tree_json_path:
+        return None
+    p = Path(tree_json_path)
+    # <session>/.coordinator/idea_tree.json -> <session>
+    return p.parent.parent if p.parent.name == ".coordinator" else p.parent
+
+
+def append_experience(tree_json_path: str | None, *, node_id: str, updates: dict[str, Any]) -> None:
+    sd = _session_dir(tree_json_path)
+    if sd is None:
+        return
+    rec = {
+        "node_id": node_id,
+        "status": updates.get("status"),
+        "insight": (updates.get("insight") or "").strip(),
+        "result": (updates.get("result") or "").strip() if updates.get("result") else "",
+        "score": updates.get("score"),
+    }
+    with open(sd / EXPERIENCE_FILENAME, "a", encoding="utf-8") as f:
+        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+
+
+def load_experience(session_dir: Path) -> list[dict[str, Any]]:
+    p = Path(session_dir) / EXPERIENCE_FILENAME
+    if not p.exists():
+        return []
+    out = []
+    for line in p.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            try:
+                out.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return out
+
+
+FINDINGS_FILENAME = "findings.jsonl"
+
+
+def record_finding(session_dir: str | Path | None, *, kind: str, about: str, note: str,
+                   source: str = "agent") -> None:
+    """Append one concrete, situational finding (a discovery or a pitfall).
+
+    These are kept specific on purpose — a dataset quirk you can exploit, a trap an
+    executor fell into — because their value is the specificity, for the next run on
+    the same/similar target. ``kind`` is free-form ('leverage' | 'pitfall' | ...).
+    """
+    if not session_dir or not (note or "").strip():
+        return
+    rec = {"kind": (kind or "").strip(), "about": (about or "").strip(),
+           "note": note.strip(), "source": source}
+    with open(Path(session_dir) / FINDINGS_FILENAME, "a", encoding="utf-8") as f:
+        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+
+
+def load_findings(session_dir: Path) -> list[dict[str, Any]]:
+    p = Path(session_dir) / FINDINGS_FILENAME
+    if not p.exists():
+        return []
+    out = []
+    for line in p.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            try:
+                out.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return out
diff --git a/src/recall.py b/src/recall.py
new file mode 100644
index 0000000..f8c37ca
--- /dev/null
+++ b/src/recall.py
@@ -0,0 +1,115 @@
+"""Experience recall: find prior runs whose lessons may help the current topic.
+
+Each finished run leaves an ``EXPERIENCE.md`` in its session folder. When a new
+run starts, intake scans past sessions, scores each against the current research
+topic, and (if a strong match exists) asks the user whether to reuse it. Accepted
+matches are composed into a tailored experience block for the agent.
+
+Matching is deterministic keyword overlap here — cheap and good enough to gate
+the ask; an embedding/LLM judge can replace ``_score`` later.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any
+
+_STOP = {"the", "a", "an", "to", "of", "and", "for", "on", "in", "with", "without",
+         "maximize", "minimize", "improve", "optimize", "score", "test", "dev", "task"}
+
+
+def _tokens(text: str) -> set[str]:
+    return {w for w in re.findall(r"[a-z0-9]+", (text or "").lower()) if len(w) > 2 and w not in _STOP}
+
+
+def _score(topic: set[str], experience: set[str]) -> float:
+    if not topic or not experience:
+        return 0.0
+    return len(topic & experience) / len(topic)  # fraction of topic terms covered
+
+
+def find_similar(cwd: str, topic: str, *, limit: int = 3, threshold: float = 0.25) -> list[dict[str, Any]]:
+    """Return prior sessions ranked by topic overlap: [{name, path, score, text}]."""
+    sessions = Path(cwd) / ".arbor" / "sessions"
+    if not sessions.is_dir():
+        return []
+    tt = _tokens(topic)
+    hits: list[dict[str, Any]] = []
+    for exp in sessions.glob("*/EXPERIENCE.md"):
+        text = exp.read_text(encoding="utf-8")
+        s = _score(tt, _tokens(text))
+        if s >= threshold:
+            hits.append({"name": exp.parent.name, "path": str(exp), "score": round(s, 3), "text": text})
+    hits.sort(key=lambda h: -h["score"])
+    return hits[:limit]
+
+
+def list_experiences(cwd: str, limit: int = 8) -> list[tuple[str, str]]:
+    """[(session_name, first-line summary)] of prior runs that left experience."""
+    sessions = Path(cwd) / ".arbor" / "sessions"
+    out: list[tuple[str, str]] = []
+    if not sessions.is_dir():
+        return out
+    for exp in sorted(sessions.glob("*/EXPERIENCE.md"), reverse=True)[:limit]:
+        desc = ""
+        for ln in exp.read_text(encoding="utf-8").splitlines():
+            if ln.startswith("description:"):
+                desc = ln.split(":", 1)[1].strip()
+                break
+        out.append((exp.parent.name, desc))
+    return out
+
+
+def compose_for_topic(cwd: str, topic: str) -> str:
+    """Compose one tailored experience block from sessions matching the topic."""
+    hits = find_similar(cwd, topic)
+    return _compose(hits)
+
+
+def compose_from_sessions(cwd: str, names: list[str]) -> str:
+    """Compose from sessions the intake agent (an LLM) judged relevant.
+
+    LLM selection beats keyword matching: the intake agent reads the goal and the
+    project, so it picks which prior runs actually transfer. Falls back to nothing
+    if the named sessions lack experience.
+    """
+    sessions = Path(cwd) / ".arbor" / "sessions"
+    hits = []
+    for name in names or []:
+        exp = sessions / name / "EXPERIENCE.md"
+        if exp.exists():
+            hits.append({"name": name, "score": "selected", "text": exp.read_text(encoding="utf-8")})
+    return _compose(hits)
+
+
+def _compose(hits: list[dict[str, Any]]) -> str:
+    """Merge findings across the matched sessions, deduped with a recurrence count.
+
+    A finding seen in several past runs is more trustworthy, so it's tagged [xN]
+    and ranked first — cross-session confidence without a global library.
+    """
+    if not hits:
+        return ""
+    merged: dict[str, list[Any]] = {}  # norm -> [count, bullet, first-session]
+    for h in hits:
+        for ln in h["text"].splitlines():
+            ln = ln.strip()
+            if not ln.startswith("- "):
+                continue
+            note = re.sub(r"\*\*\[.*?\]\s*[^—]*\*\*\s*—?\s*", "", ln[2:]).strip()
+            key = re.sub(r"\W+", "", note.lower())[:80]
+            if not key:
+                continue
+            if key in merged:
+                merged[key][0] += 1
+            else:
+                merged[key] = [1, ln[2:].strip(), h["name"]]
+    if not merged:  # nothing parseable — fall back to raw concatenation
+        return "\n".join(["# Prior experience (candidate priors — verify, don't blindly apply)"]
+                         + [f"\n## from {h['name']}\n{h['text']}" for h in hits])
+    ranked = sorted(merged.values(), key=lambda x: -x[0])
+    lines = ["# Prior experience (candidate priors — verify, don't blindly apply)",
+             f"_merged from {len(hits)} past run(s); [xN] = seen in N runs_\n"]
+    lines += [f"- [x{c}] {bullet}" for c, bullet, _src in ranked]
+    return "\n".join(lines)