feat(experience): self-evolution line 2 — capture & reuse concrete findings

src/cli/intake/launch_tool.py

@@ -129,6 +129,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"],
     }
@@ -165,7 +176,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")),
@@ -188,3 +201,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

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 "

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)

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
     # ------------------------------------------------------------------

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)

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]}"

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}")

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