From 46dcf6783c9d1466e5039651b0c84766ae5ad653 Mon Sep 17 00:00:00 2001 From: ignorejjj Date: Mon, 29 Jun 2026 13:16:05 +0800 Subject: [PATCH 1/4] docs(dev): spec trajectory export for self-evolution (SFT/RL + skills, Polar-aware) --- docs/dev/trajectory-export.md | 92 +++++++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 docs/dev/trajectory-export.md diff --git a/docs/dev/trajectory-export.md b/docs/dev/trajectory-export.md new file mode 100644 index 0000000..2588fa1 --- /dev/null +++ b/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 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. From d7c01ee42e0994ebbc43b3e2ff11ffd03562e570 Mon Sep 17 00:00:00 2001 From: ignorejjj Date: Mon, 29 Jun 2026 13:22:53 +0800 Subject: [PATCH 2/4] feat(trajectory): export training-ready decision trace at finalize Joins events.jsonl + idea tree into trajectory.jsonl (one record per idea: hypothesis, reward from eval, status, insight). Auto-written at finalize (config evolution flag export_trajectory, default on) and re-exportable for old runs via 'arbor export --format trajectory'. Pure transform, no API. --- src/cli/commands/export_cmd.py | 7 +- src/coordinator/config.py | 4 ++ src/coordinator/orchestrator.py | 10 +++ src/trajectory.py | 116 ++++++++++++++++++++++++++++++++ 4 files changed, 136 insertions(+), 1 deletion(-) create mode 100644 src/trajectory.py diff --git a/src/cli/commands/export_cmd.py b/src/cli/commands/export_cmd.py index fcb5b7b..8c3bc78 100644 --- a/src/cli/commands/export_cmd.py +++ b/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) diff --git a/src/coordinator/config.py b/src/coordinator/config.py index d9f415b..1cb24b1 100644 --- a/src/coordinator/config.py +++ b/src/coordinator/config.py @@ -471,6 +471,10 @@ 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 + # ── 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 f5f4c95..7185dff 100644 --- a/src/coordinator/orchestrator.py +++ b/src/coordinator/orchestrator.py @@ -1139,6 +1139,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 # ------------------------------------------------------------------ diff --git a/src/trajectory.py b/src/trajectory.py new file mode 100644 index 0000000..5b0dfe9 --- /dev/null +++ b/src/trajectory.py @@ -0,0 +1,116 @@ +"""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)) From f6a43830cbaefee4a506e9dce6026678bda2dce6 Mon Sep 17 00:00:00 2001 From: ignorejjj Date: Mon, 29 Jun 2026 13:44:58 +0800 Subject: [PATCH 3/4] feat(trajectory): opt-in token-level call trace (tokens.jsonl) for SFT/RL Each coordinator LLM call appends messages+output+usage to tokens.jsonl when coordinator.token_trace is on (default off, heavy). logprobs/token_ids recorded only when the provider returns them (OpenAI); Anthropic gives none, stored null rather than faked. Polar-shaped; no provider rewrite, no proxy. --- src/coordinator/config.py | 2 ++ src/coordinator/orchestrator.py | 5 +++++ src/core/agent.py | 9 +++++++++ src/core/config.py | 4 ++++ src/trajectory.py | 31 +++++++++++++++++++++++++++++++ 5 files changed, 51 insertions(+) diff --git a/src/coordinator/config.py b/src/coordinator/config.py index 1cb24b1..c1fc22d 100644 --- a/src/coordinator/config.py +++ b/src/coordinator/config.py @@ -474,6 +474,8 @@ class CoordinatorConfig(ProxyModel): # ── 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) diff --git a/src/coordinator/orchestrator.py b/src/coordinator/orchestrator.py index 7185dff..0e47377 100644 --- a/src/coordinator/orchestrator.py +++ b/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 diff --git a/src/core/agent.py b/src/core/agent.py index bf947be..116bb26 100644 --- a/src/core/agent.py +++ b/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}) diff --git a/src/core/config.py b/src/core/config.py index 5abd3a5..2ed2f72 100644 --- a/src/core/config.py +++ b/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 diff --git a/src/trajectory.py b/src/trajectory.py index 5b0dfe9..d29eeb6 100644 --- a/src/trajectory.py +++ b/src/trajectory.py @@ -114,3 +114,34 @@ def write_trajectory(session_dir: Path) -> Path: 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 From bbd01dcff486a27fc2ef724e01baf99b3043e7cf Mon Sep 17 00:00:00 2001 From: ignorejjj Date: Mon, 29 Jun 2026 13:49:13 +0800 Subject: [PATCH 4/4] feat(llm): plumb sampled-token logprobs into LLMResponse (chat path) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add LLMResponse.logprobs; request logprobs=true on the openai-compat chat call and parse choice.logprobs into [{token,logprob}]. Endpoints without logprobs (4141/Claude today) return None — interface is ready so token_trace becomes RL-grade the moment a logprob-capable endpoint is used. --- src/core/llm/base.py | 4 ++++ src/core/llm/openai_compat.py | 20 ++++++++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/src/core/llm/base.py b/src/core/llm/base.py index 7c07cc4..d3c51ba 100644 --- a/src/core/llm/base.py +++ b/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)) diff --git a/src/core/llm/openai_compat.py b/src/core/llm/openai_compat.py index a08bee9..44c213e 100644 --- a/src/core/llm/openai_compat.py +++ b/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), )