feat(trajectory): self-evolution line 1 — training-ready run traces

docs/dev/trajectory-export.md

@@ -0,0 +1,92 @@
+# Trajectory Export (self-evolution · line 1)
+
+Status: spec for review. No code yet.
+
+## Goal
+
+Every finished run already writes `events.jsonl`. This adds a clean,
+training-ready dump so users can collect runs for SFT/RL. No new top-level
+command — it rides the existing finalize step, like `REPORT.md` does.
+
+## What the user sees
+
+After a run, the session dir gains one file: `trajectory.jsonl`. That's it. A
+config flag turns it off:
+
+```yaml
+evolution:
+  export_trajectory: true   # default on; writes trajectory.jsonl at finalize
+```
+
+No new verb, no extra prompt. Collecting many runs = globbing
+`.arbor/sessions/*/trajectory.jsonl`.
+
+## What it is — two granularities
+
+Prior art: NVIDIA Polar (arXiv 2605.24220, "Agentic RL on Any Harness at
+Scale"). Polar proxies the LLM API boundary, captures token-faithful traces
+(messages, token_ids, logprobs, loss mask, reward) and feeds async RL trainers.
+Lesson: for RL you need token-level fidelity, not coarse decision logs. Arbor
+needs **no proxy** — it owns its provider layer, so it records the same thing
+directly at the call site.
+
+- **Coarse trace** (`trajectory.jsonl`): one line per decision point — for
+  analysis and SFT of the coordinator policy. Cheap, always on.
+- **Token-faithful trace** (`tokens.jsonl`, opt-in): one record per LLM call —
+  messages, sampled token_ids, logprobs, loss mask (only behavior-policy tokens
+  trainable), reward back-filled from eval. RL-ready, Polar-compatible. Needs a
+  provider that returns logprobs/token_ids (OpenAI yes; Anthropic limited — note
+  the gap, don't fake it).
+
+## Where the data comes from
+
+All of it already exists; this just joins it up. Source = `events.jsonl` +
+`.coordinator/idea_tree.json`:
+
+| Record field | Source event |
+| --- | --- |
+| `cycle`, `phase` | `cycle.start` / `cycle.phase` |
+| `proposed` (node_id, hypothesis, parent) | `idea.proposed` |
+| `executor` action (branch, code edits) | `executor.start` / `executor.end` |
+| `reward` (dev score, delta) | `eval.end` / `idea.completed` |
+| `decision` (merge/prune/stop) | `idea.merged` / `idea.pruned` |
+| outcome status, final insight | idea_tree node |
+
+## Record shape (one line)
+
+```json
+{
+  "run": "run_20260628_230556",
+  "step": 4,
+  "cycle": 2,
+  "node_id": "2.1",
+  "parent_id": "2",
+  "state": {"frontier": [...], "best_score": 2.16, "constraints": "..."},
+  "action": {"kind": "ideate", "hypothesis": "GEMM-expanded squared distances + argpartition"},
+  "reward": {"dev_score": 7.50, "delta": 5.34, "merged": true},
+  "tokens": 16512
+}
+```
+
+State is the tree digest the coordinator was grounded on (we already build this
+for IDEATE — reuse it). Reward fills in when that node's `eval.end` lands;
+non-scored steps get `reward: null`.
+
+## How it's built
+
+1. Reuse `export.py`'s session resolver + jsonl reader (don't rebuild).
+2. Replay `events.jsonl` in order, carry a small state, emit one record per
+   decision point, back-fill reward when eval ends.
+3. Write `trajectory.jsonl`. Hook = same finalize path that writes `run_stats.json`.
+4. `arbor export <session> traj.jsonl` keeps working for old runs (offline).
+
+## Validate (no API needed)
+
+Run on the existing `/tmp/algotune_knn_dogfood` session: confirm ~6 ideas →
+records, scores back-filled, file loads in a training loop. Pure transform, no
+model calls — cheap, deterministic.
+
+## Out of scope (line 2)
+
+Distilled skills, preferences, novelty ledger, recall. This file is only the raw
+training dump. Skills sit next to it later, same finalize hook.

src/cli/commands/export_cmd.py

@@ -24,7 +24,7 @@ def export_command(
         None,
         "--format",
         "-f",
-        help="Export format: html or jsonl. Inferred from output extension when omitted.",
+        help="Export format: html, jsonl, or trajectory. Inferred from output extension when omitted.",
     ),
 ) -> None:
     """Export a previous Arbor session for review or sharing."""
@@ -33,6 +33,11 @@ def export_command(
 
     try:
         session_dir = resolve_session_dir(session, cwd)
+        if (fmt or "").lower() == "trajectory":
+            from ...trajectory import write_trajectory
+            path = write_trajectory(session_dir)
+            typer.secho(f"Exported trajectory to: {path}", fg=typer.colors.GREEN)
+            return
         result = export_session(session_dir, output, fmt=fmt)
     except ExportError as exc:
         typer.secho(f"error: {exc}", fg=typer.colors.RED, err=True)

src/coordinator/config.py

@@ -471,6 +471,12 @@ class CoordinatorConfig(ProxyModel):
     skills_enabled: bool = True
     disabled_skills: list[str] = PydField(default_factory=list)
 
+    # ── Self-evolution ───────────────────────────────────────────────
+    # Write a training-ready decision trace (trajectory.jsonl) at finalize.
+    export_trajectory: bool = True
+    # Capture per-call token-level traces (tokens.jsonl) for SFT/RL. Heavy; off.
+    token_trace: bool = False
+
     # ── Plugin (runtime object; not serialized) ──────────────────────
     plugin: Any = PydField(default=None, exclude=True, repr=False)
 

src/coordinator/orchestrator.py

@@ -630,6 +630,11 @@ def _create_coordinator(self) -> Agent:
             verbose=self.config.verbose,
             workspace_dir=self.config.workspace_dir,
             agent_label="coordinator",
+            token_trace_path=(
+                str(Path(self.config.workspace_dir) / "tokens.jsonl")
+                if getattr(self.config, "token_trace", False) and self.config.workspace_dir
+                else None
+            ),
             inter_turn_user_messages=_drain_dashboard_messages,
             checkpoint_hook=lambda msgs, turn: self._write_checkpoint(
                 reason="turn", messages=msgs
@@ -1139,6 +1144,16 @@ def _write_run_stats(
         except OSError as e:
             _print_status(f"Warning: failed to write run_stats.json: {e}")
 
+        # Self-evolution line 1: dump a training-ready decision trace alongside the
+        # stats. Best-effort — never fail a finished run over an export. Off via
+        # `evolution.export_trajectory: false`.
+        if getattr(self.config, "export_trajectory", True):
+            try:
+                from ..trajectory import write_trajectory
+                _print_status(f"Wrote trajectory: {write_trajectory(self.config.workspace_dir)}")
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                _print_status(f"Warning: failed to write trajectory: {e}")
+
     # ------------------------------------------------------------------
     # Final report
     # ------------------------------------------------------------------

src/core/agent.py

@@ -352,6 +352,15 @@ async def _run_loop(self, user_message: str) -> str:
                 agent_cwd=self.config.cwd,
                 track_stats=self.config.track_stats,
             )
+            if self.config.token_trace_path:
+                from ..trajectory import append_token_record
+                append_token_record(
+                    self.config.token_trace_path,
+                    messages=self.messages,
+                    response=response,
+                    turn=turn,
+                    model=self.provider.model,
+                )
 
             # 3. Append assistant message
             self.messages.append({"role": "assistant", "content": response.raw_content})

src/core/config.py

@@ -79,6 +79,10 @@ class AgentConfig(ProxyModel):
     # run's stats (#13 cache_hit_rate, the dashboard token counter, run_stats).
     track_stats: bool = Field(default=True, exclude=True, repr=False)
 
+    # Token-level trace sink (self-evolution line 1). When set, each LLM call's
+    # messages + output are appended to this jsonl for SFT/RL. None = off.
+    token_trace_path: str | None = Field(default=None, exclude=True, repr=False)
+
     # ── Derived paths ────────────────────────────────────────────────
 
     @property

src/core/llm/base.py

@@ -83,6 +83,10 @@ class LLMResponse:
     # Populated by the provider so that appending to messages[] is lossless.
     raw_content: list[dict[str, Any]] = field(default_factory=list)
 
+    # Sampled-token logprobs when the provider returns them (OpenAI logprobs=true);
+    # None otherwise (e.g. Anthropic). For token-faithful SFT/RL traces.
+    logprobs: list[dict[str, Any]] | None = None
+
     def get_text(self) -> str:
         """Concatenate all text blocks."""
         return "\n".join(b.text for b in self.content if isinstance(b, TextBlock))

src/core/llm/openai_compat.py

@@ -45,6 +45,22 @@ def _cached_prompt_tokens(usage: Any) -> int:
     return int(cached or 0)
 
 
+def _extract_logprobs(choice: Any) -> list[dict[str, Any]] | None:
+    """Pull sampled-token logprobs from a chat choice, or None if absent.
+
+    Returns ``[{token, logprob}, ...]`` for token-faithful traces. Endpoints
+    that don't return logprobs (most non-OpenAI gateways) yield None.
+    """
+    lp = getattr(choice, "logprobs", None)
+    content = getattr(lp, "content", None) if lp is not None else None
+    if not content:
+        return None
+    out: list[dict[str, Any]] = []
+    for tok in content:
+        out.append({"token": getattr(tok, "token", None), "logprob": getattr(tok, "logprob", None)})
+    return out or None
+
+
 class OpenAICompatProvider(LLMProvider):
     """LLM provider for any OpenAI-compatible API.
 
@@ -107,6 +123,9 @@ async def create(
             "model": self.model,
             "messages": oai_messages,
             "max_tokens": max_tokens,
+            # Token-faithful traces: ask for sampled-token logprobs. Endpoints that
+            # don't support it ignore these; we read back None and degrade cleanly.
+            "logprobs": True,
         }
         if oai_tools:
             params["tools"] = oai_tools
@@ -405,4 +424,5 @@ def _field(name: str) -> Any:
             usage=usage,
             model=raw.model or self.model,
             raw_content=raw_content,
+            logprobs=_extract_logprobs(choice),
         )

src/trajectory.py

@@ -0,0 +1,147 @@
+"""Coarse trajectory export for self-evolution (line 1).
+
+Joins ``events.jsonl`` + the idea tree into one ``trajectory.jsonl`` — one line
+per decision point (a proposed idea, its execution, and the merge/prune
+decision), reward back-filled from eval. Pure transform: no LLM calls, no
+network. Token-faithful per-call traces (Polar-style) are a separate, opt-in
+layer; see ``docs/dev/trajectory-export.md``.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from .export import resolve_session_dir  # reuse the existing resolver
+
+TRAJECTORY_FILENAME = "trajectory.jsonl"
+
+
+def _load_jsonl(path: Path) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    if not path.exists():
+        return out
+    for line in path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            out.append(json.loads(line))
+        except json.JSONDecodeError:
+            continue
+    return out
+
+
+def _load_tree(session_dir: Path) -> dict[str, dict[str, Any]]:
+    for rel in (".coordinator/idea_tree.json", "idea_tree.json"):
+        p = session_dir / rel
+        if p.exists():
+            try:
+                nodes = json.loads(p.read_text(encoding="utf-8")).get("nodes", {})
+            except json.JSONDecodeError:
+                return {}
+            return nodes if isinstance(nodes, dict) else {n.get("id"): n for n in nodes}
+    return {}
+
+
+def build_trajectory(session_dir: Path) -> list[dict[str, Any]]:
+    """Reconstruct ordered decision-point records for one session."""
+    session_dir = Path(session_dir)
+    events = _load_jsonl(session_dir / "events.jsonl")
+    tree = _load_tree(session_dir)
+    run = session_dir.name
+
+    cycle_of: dict[str, int] = {}
+    cur_cycle = 0
+    proposed: dict[str, dict[str, Any]] = {}      # node_id -> action payload
+    rewards: dict[str, dict[str, Any]] = {}        # node_id -> reward
+    order: list[str] = []
+
+    for ev in events:
+        t, d = ev.get("type"), ev.get("data", {})
+        if t == "cycle.start":
+            cur_cycle = d.get("cycle_num", cur_cycle)
+            nid = d.get("node_id")
+            if nid:
+                cycle_of[nid] = cur_cycle
+        elif t == "idea.proposed":
+            nid = d.get("node_id")
+            if nid and nid not in proposed:
+                proposed[nid] = {"hypothesis": d.get("hypothesis", ""), "parent_id": d.get("parent_id")}
+                cycle_of.setdefault(nid, cur_cycle)
+                order.append(nid)
+        elif t in ("eval.end", "executor.end", "idea.completed"):
+            nid = d.get("node_id")
+            if nid:
+                r = rewards.setdefault(nid, {})
+                if d.get("score") is not None:
+                    r["dev_score"] = d["score"]
+                r.setdefault("merged", False)
+                if d.get("tokens") is not None:
+                    r["tokens"] = d["tokens"]
+        elif t == "idea.merged":
+            nid = d.get("node_id")
+            if nid:
+                rewards.setdefault(nid, {})["merged"] = True
+
+    records: list[dict[str, Any]] = []
+    for step, nid in enumerate(order):
+        node = tree.get(nid, {})
+        records.append({
+            "run": run,
+            "step": step,
+            "node_id": nid,
+            "parent_id": proposed[nid].get("parent_id"),
+            "cycle": cycle_of.get(nid, 0),
+            "action": {"kind": "ideate", "hypothesis": proposed[nid]["hypothesis"]},
+            "reward": rewards.get(nid) or None,
+            "outcome": {"status": node.get("status"), "score": node.get("score")},
+            "insight": node.get("insight", ""),
+        })
+    return records
+
+
+def write_trajectory(session_dir: Path) -> Path:
+    """Write ``trajectory.jsonl`` into the session dir; return its path."""
+    session_dir = Path(session_dir)
+    records = build_trajectory(session_dir)
+    out = session_dir / TRAJECTORY_FILENAME
+    out.write_text("\n".join(json.dumps(r, ensure_ascii=False) for r in records) + "\n", encoding="utf-8")
+    return out
+
+
+def export_trajectory(session: str, cwd: Path | None = None) -> Path:
+    """Resolve a session name/path and write its trajectory (CLI/offline use)."""
+    return write_trajectory(resolve_session_dir(Path(session), cwd))
+
+
+def append_token_record(
+    path: str | Path,
+    *,
+    messages: list[dict[str, Any]],
+    response: Any,
+    turn: int,
+    model: str,
+) -> None:
+    """Append one token-level call record (Polar-style) to ``tokens.jsonl``.
+
+    Captures the request context and the model output for SFT. logprobs/token_ids
+    are filled only when the provider returns them (OpenAI); Anthropic gives
+    none, so we record null rather than fake it. Best-effort — never raise.
+    """
+    try:
+        u = getattr(response, "usage", None)
+        rec = {
+            "turn": turn,
+            "model": model,
+            "messages": messages,
+            "output": getattr(response, "raw_content", None),
+            "input_tokens": int(getattr(u, "input_tokens", 0) or 0),
+            "output_tokens": int(getattr(u, "output_tokens", 0) or 0),
+            "logprobs": getattr(response, "logprobs", None),  # None unless provider gives it
+        }
+        with open(path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+    except Exception:  # pylint: disable=broad-exception-caught
+        pass