feat(arms): multi-turn session measurement with per-turn usage

dokimi/changelog.d/20261021_130000_napetrov_skill-gap-W7.md

@@ -0,0 +1,8 @@
+<!--
+Copyright (C) 2026 Intel Corporation
+SPDX-License-Identifier: Apache-2.0
+-->
+
+### Added
+
+- **Multi-turn session measurement for `arms run`** — `--session-turns N` asks each question as its own text followed by the first N-1 entries of its `follow_ups` list, keeps the conversation history across the turns, and records each turn's input, output, cache read and write tokens and cost under `session.turns` of the arm. The arm's `token_usage` and `metrics` are the sums over turns, so cost rollups read them as before. `--turn-reminder TEXT` also runs each arm as `<arm>+reminder`, with the text appended to every user turn in a `<system-reminder>` block, to test whether a per-turn reminder keeps an instruction in force in later turns. Measurement only: `--judge` is refused with `--session-turns` and every `eval` judge command refuses a session artifact, a question with too few `follow_ups` fails the run before any answer is paid for, and agentic arms (`mcp:`, `skill-agent:`, `skill-catalog:`) are refused per arm. Without the flags, `arms run` behaves as before.

dokimi/docs/benchmarking-and-comparison.md

@@ -383,6 +383,39 @@ that needed a re-draw.
 token, but some endpoints compute slightly different scores from one call to
 the next, so reproducibility has to be measured per endpoint.
 
+### Multi-turn sessions
+
+A single answer does not show what an arm costs once the conversation grows.
+`--session-turns N` asks each question as its own text followed by the first
+N-1 entries of its `follow_ups` list (a list of strings in the question record),
+keeps the history, and records each turn's usage:
+
+```bash
+python cli.py arms run ... --arms baseline,skill:/path/to/skill --session-turns 4
+python cli.py arms run ... --arms baseline,skill:/path/to/skill --session-turns 4 \
+  --turn-reminder "Answer tersely."   # also runs each arm as <arm>+reminder
+```
+
+Every turn is recorded under the arm's `session.turns`, with `metrics` for that
+call (prompt, completion, cache read and write tokens, cost). The arm's
+`token_usage` and `metrics` are the sums over turns, so `cost_summary` reads
+them like any other run. A question with fewer than N-1 follow-ups fails the
+run before any answer is generated. Agentic arms (`mcp:`, `skill-agent:`,
+`skill-catalog:`) are refused per arm, and a session cannot be judged:
+`--judge` is refused with `--session-turns`, and every `eval` judge command
+refuses a session artifact, because the judge would score the last answer
+against the first question. The reminder is appended to every user
+turn in a `<system-reminder>` block; the assistant turns are not changed.
+
+A session that fails part-way (a turn raises, or a turn comes back empty and the
+session stops) is an errored arm: `error` is `generation_failed: turn N: ...` or
+`empty_reply: turn N` (turns count from 0), and `answer` is `None`. The turns already paid for stay in
+`session.turns`, and the arm's `token_usage` and `metrics` cover them, so the cost
+rollup counts what was spent. The artifact's `run_config` records `session_turns`
+and `turn_reminder`. `plugin-delta` refuses to pair a session run with a
+single-turn run or one with different session settings, and the matrix rollup
+does not pair them.
+
 ### Trusting the judge
 
 A significance table says the delta is unlikely under *this judge's* noise. It

dokimi/dokimi/commands/arms.py

@@ -46,6 +46,8 @@
     "judge_panel",
     "concurrency",
     "stats_correction",
+    "session_turns",
+    "turn_reminder",
 )
 
 
@@ -289,6 +291,14 @@ def register(sub, positive_int) -> None:
                                  "'technical_expert,developer_advocate,doc_reviewer'). When "
                                  "set, --judge scores each answer with a panel and records "
                                  "grader spread and agreement instead of one opinion.")
+    arms_run_p.add_argument("--session-turns", type=positive_int, default=None, dest="session_turns",
+                            help="Measure multi-turn sessions: each question is asked as its question "
+                                 "followed by the first N-1 entries of its 'follow_ups' list, with the "
+                                 "history kept, and each turn's token and cache usage recorded. Non-agentic "
+                                 "arms only. Measurement only: refused with --judge")
+    arms_run_p.add_argument("--turn-reminder", default=None, dest="turn_reminder",
+                            help="With --session-turns, also run each arm as '<arm>+reminder', with this "
+                                 "text appended to every user turn in a <system-reminder> block")
     arms_run_p.add_argument("--concurrency", type=positive_int, default=5)
     arms_run_p.add_argument("--out-json", default=None, dest="out_json",
                             help="Output JSON path (default: results/arms/{product}.json)")
@@ -562,6 +572,12 @@ def _probe_sampling(model: str, provider: str, temperature, seed) -> None:
         ) from exc
 
 
+def _refuse_judged_session(args: argparse.Namespace) -> None:
+    """Refused before any answer is paid for: a session is measured, not judged."""
+    if getattr(args, "session_turns", None) is not None and args.judge:
+        raise ValueError("--session-turns is measurement-only; drop --judge")
+
+
 def _build_arms_output(args: argparse.Namespace, specs: list[str], questions: list[dict]) -> dict:
     from dokimi.eval import Judge, JudgePanel
     from dokimi.eval.panel import JUDGE_ROLES, JudgeConfig
@@ -574,6 +590,7 @@ def _build_arms_output(args: argparse.Namespace, specs: list[str], questions: li
     )
     from dokimi.treatments import create_treatments
 
+    _refuse_judged_session(args)
     treatments = create_treatments(specs, top_k=args.top_k, rerank_threshold=args.rerank_threshold)
     _validate_expected_skills(questions, treatments)
     try:
@@ -657,6 +674,8 @@ def _build_arms_output(args: argparse.Namespace, specs: list[str], questions: li
         stats_correction=getattr(args, "stats_correction", "holm"),
         temperature=getattr(args, "temperature", None),
         seed=getattr(args, "seed", None),
+        session_turns=getattr(args, "session_turns", None),
+        turn_reminder=getattr(args, "turn_reminder", None),
     )
     records = runner.run(
         library_name=args.product,

dokimi/dokimi/eval/arm_runner.py

@@ -37,6 +37,7 @@
 from dokimi.eval.judge_context import build as build_judge_context
 from dokimi.eval.judge_context import cite as cite_judge_materials
 from dokimi.eval.judge_context import join_injected
+from dokimi.eval.session import REMINDER_ARM_SUFFIX, run_session, session_turns_of
 from dokimi.metrics.usage import to_record
 from dokimi.treatments.base import Treatment
 
@@ -109,6 +110,16 @@ def _static_arm_kind(treatment: Treatment) -> Dict[str, Any]:
     return {"arm_kind": getattr(t, "arm_kind", None)}
 
 
+
+def _refuse_session_records(recs: Sequence[Record]) -> None:
+    """Refuse to judge a session run: its last answer would be scored against its first question."""
+    if any(r.raw.get("session_turns") for r in recs):
+        raise ValueError(
+            "Session records are measurement-only: the judge would score a session's last "
+            "answer against its first question. Judge a single-turn run instead."
+        )
+
+
 class ArmRunner:
     """Generate answers for each treatment arm and (optionally) judge them."""
 
@@ -133,9 +144,17 @@ def __init__(
         stats_correction: str = "holm",
         temperature: Optional[float] = None,
         seed: Optional[int] = None,
+        session_turns: Optional[int] = None,
+        turn_reminder: Optional[str] = None,
     ):
         if not treatments:
             raise ValueError("ArmRunner requires at least one treatment arm")
+        if session_turns is not None and session_turns < 2:
+            raise ValueError(
+                f"session_turns must be at least 2 (one turn is the plain run), got {session_turns}"
+            )
+        if turn_reminder is not None and (session_turns is None or not turn_reminder.strip()):
+            raise ValueError("turn_reminder needs session_turns and non-blank text")
         # Rejected here rather than at summarize time: the correction is only
         # applied after every arm has been answered and judged, so a typo would
         # otherwise surface after the whole run has been paid for.
@@ -153,6 +172,8 @@ def __init__(
         self.max_tokens = max_tokens
         self.temperature = temperature
         self.seed = seed
+        self.session_turns = session_turns
+        self.turn_reminder = turn_reminder
         self.grounded_from_topics = grounded_from_topics
         self.harness = harness
         self.matrix_cell = matrix_cell
@@ -168,7 +189,10 @@ def __init__(
 
     @property
     def arm_names(self) -> List[str]:
-        return [t.name for t in self.treatments]
+        names = [t.name for t in self.treatments]
+        if self.turn_reminder is not None:
+            names += [name + REMINDER_ARM_SUFFIX for name in names]
+        return names
 
     def _run_config(self) -> Dict[str, Any]:
         """Request-shaping settings, recorded so a run stays interpretable later.
@@ -183,13 +207,19 @@ def _run_config(self) -> Dict[str, Any]:
         non-deterministic by default, so a pinned run and a free-sampling one are
         differently reproducible and should not be silently pooled as replicates.
         """
-        return {
+        config: Dict[str, Any] = {
             "max_tokens": self.max_tokens,
             "max_iterations": self.max_iterations,
             "grounded_from_topics": self.grounded_from_topics,
             "temperature": self.temperature,
             "seed": self.seed,
         }
+        # A session run and a single-turn run answer different questions of the model (the
+        # history grows), so the pairing checks must be able to tell them apart.
+        if self.session_turns is not None:
+            config["session_turns"] = self.session_turns
+            config["turn_reminder"] = self.turn_reminder
+        return config
 
     def run(
         self,
@@ -200,6 +230,12 @@ def run(
     ) -> List[Dict[str, Any]]:
         """Answer every question under every arm. Returns one record per question."""
         n = len(questions)
+        if self.session_turns is not None:
+            # Before any answer is paid for: a question short of follow-ups fails the run.
+            for i, q in enumerate(questions):
+                qid = question_id(q)
+                session_turns_of(q, question_text_of(q), self.session_turns,
+                                 label=f"#{i}" if qid is None else str(qid))
         results: Dict[int, Dict[str, Any]] = {}
         lock = threading.Lock()
         done = [0]
@@ -230,11 +266,25 @@ def _answer_question(
         if qid is None:
             qid = "unknown_q"
 
+        turns = None
+        if self.session_turns is not None:
+            turns = session_turns_of(q, question_text, self.session_turns, label=qid)
+
         arms: Dict[str, Any] = {}
         for treatment in self.treatments:
-            arms[treatment.name] = self._answer_one_arm(
-                treatment, question_text, library_name, library_id, qid
+            if turns is None:
+                arms[treatment.name] = self._answer_one_arm(
+                    treatment, question_text, library_name, library_id, qid
+                )
+                continue
+            arms[treatment.name] = self._answer_session_arm(
+                treatment, turns, library_name, library_id, qid
             )
+            if self.turn_reminder is not None:
+                arms[treatment.name + REMINDER_ARM_SUFFIX] = self._answer_session_arm(
+                    treatment, turns, library_name, library_id, qid,
+                    reminder=self.turn_reminder,
+                )
 
         rec = {
             "question_id": qid,
@@ -245,6 +295,8 @@ def _answer_question(
             "persona": q.get("persona"),
             "arms": arms,
         }
+        if turns is not None:
+            rec["session_turns"] = len(turns)
         # Selection label: present-and-None means "no skill should load", which is
         # why presence is tested rather than truthiness.
         if "expected_skill" in q:
@@ -365,6 +417,114 @@ def _answer_one_arm(
             "elapsed_sec": round(time.time() - t0, 2),
         }
 
+    def _answer_session_arm(
+        self,
+        treatment: Treatment,
+        turns: List[str],
+        library_name: str,
+        library_id: Optional[str],
+        question_id: str,
+        reminder: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Answer one arm's multi-turn session; see :mod:`dokimi.eval.session`."""
+        t0 = time.time()
+        try:
+            cfg = treatment.prepare(turns[0], library_name, library_id)
+        except Exception as exc:
+            logger.exception("Arm %s prepare() failed", treatment.name)
+            return {"error": f"prepare_failed: {exc}", "answer": None,
+                    "metadata": _static_arm_kind(treatment)}
+
+        from dokimi.plugins import take_output_shapers
+        shapers, cfg_metadata = take_output_shapers(cfg)
+        if cfg.is_agentic:
+            return {"error": "session_unsupported: an agentic arm (tools) is answered in one "
+                             "turn only", "answer": None, "metadata": cfg_metadata}
+
+        # The opening message is the single-turn prompt, context included, so turn 1 of
+        # a session is the answer the plain run would have given, cache breakpoint included.
+        cache_prefix = None
+        if cfg.has_context:
+            context = join_injected(cfg.injected_context)
+            prefix = CONTEXT_PROMPT_PREFIX.format(context=context[:MAX_CONTEXT_CHARS])
+            opening = CONTEXT_PROMPT_SUFFIX.format(question=turns[0])
+            if cfg.context_is_static:
+                cache_prefix = prefix
+            else:
+                opening = prefix + opening
+        else:
+            opening = ANSWER_PROMPT_PLAIN.format(question=turns[0])
+
+        session = run_session(
+            model=self.model, provider=self.provider, api_key=self.api_key,
+            opening=opening, follow_ups=turns[1:], system=cfg.system_prompt,
+            max_tokens=self.max_tokens, temperature=self.temperature, seed=self.seed,
+            reminder=reminder, cache_prefix=cache_prefix,
+        )
+        session_block = {
+            "turns_requested": len(turns),
+            "turns": session["turns"],
+            "stopped_reason": session["stopped_reason"],
+            "reminder": reminder is not None,
+        }
+        if session["error"] is not None:
+            logger.error("Arm %s session failed for %s: %s", treatment.name,
+                         question_id, session["error"])
+            return {"error": f"generation_failed: {session['error']}", "answer": None,
+                    **self._session_spend(session),
+                    "session": session_block, "metadata": cfg_metadata}
+
+        if session["stopped_reason"] == "empty_reply":
+            # The session ended short of the turns asked for. Recording it as a success
+            # would mix 2-turn and 3-turn sessions in one arm's totals; the usage stays.
+            error = f"empty_reply: turn {len(session['turns']) - 1}"
+            logger.error("Arm %s session stopped for %s: %s", treatment.name, question_id, error)
+            return {"error": error, "answer": None, **self._session_spend(session),
+                    "session": session_block, "metadata": cfg_metadata}
+
+        raw_answer = session["answer"]
+        final_text = self._apply_shapers(shapers, raw_answer)
+        usage = session["usage"]
+        return {
+            "answer": final_text,
+            "raw_answer": raw_answer if shapers else None,
+            "model": self.model,
+            "harness": self.harness,
+            "plugin_set": self.plugin_set["plugin_set"],
+            "plugin_set_id": self.plugin_set["plugin_set_id"],
+            "used_system_prompt": bool(cfg.system_prompt),
+            "context_chunks": [
+                {
+                    "source": c.get("source", "unknown"),
+                    "snippet": (c["content"][:200] + "…") if len(c["content"]) > 200 else c["content"],
+                    "relevance_score": c.get("relevance_score"),
+                }
+                for c in cfg.injected_context
+            ],
+            # Whole-session totals, so the cost and token rollups read this arm like any other.
+            "token_usage": usage.as_token_usage_dict(),
+            "metrics": usage.as_metrics_dict(
+                answer_chars=len(raw_answer or ""), final_chars=len(final_text or "")),
+            "session": session_block,
+            "metadata": cfg_metadata,
+            "elapsed_sec": round(time.time() - t0, 2),
+        }
+
+    @staticmethod
+    def _session_spend(session: Dict[str, Any]) -> Dict[str, Any]:
+        """Token usage and metrics of the turns a failed session did complete.
+
+        The calls were paid for, so the cost rollup must count them; an arm without
+        ``metrics`` is skipped by it. No answer-length keys are written: a failed
+        session has no answer, and plugin-delta averages those keys over every arm
+        that has a ``metrics`` block.
+        """
+        if not session["turns"]:
+            return {}
+        usage = session["usage"]
+        return {"token_usage": usage.as_token_usage_dict(),
+                "metrics": usage.as_metrics_dict()}
+
     @staticmethod
     def _apply_shapers(shapers, answer_text):
         """Run output shapers over a raw answer (no-op when there are none)."""
@@ -497,6 +657,7 @@ def _score(rec: Record) -> Dict[str, Any]:
         recs = as_records(records, source="answers", kind="answers")
         # Before any judge call: refuse, do not grade against "".
         require_question_texts(r.raw for r in recs)
+        _refuse_session_records(recs)
         n = len(recs)
         out: Dict[int, Dict[str, Any]] = {}
         with ThreadPoolExecutor(max_workers=concurrency) as pool:
@@ -539,6 +700,8 @@ def build_output(
             "agentic_usage_summary": self._summarize_agentic_usage(records),
             "skill_selection_summary": summarize_skill_selection(records),
         }
+        if self.session_turns is not None:
+            out["session"] = {"turns": self.session_turns, "turn_reminder": self.turn_reminder}
         # One copy of each judge context the arm records cite, written whether or
         # not the run was judged, so an answers-only artifact can be judged later.
         materials = cite_judge_materials(records, self.judge_context)

dokimi/dokimi/eval/matrix_rollup.py

@@ -124,6 +124,8 @@ def _pair_key(data: dict[str, Any]) -> tuple[Any, ...]:
         tuple(data.get("arms") or []),
         data.get("baseline_arm"),
         data.get("total_questions"),
+        (data.get("run_config") or {}).get("session_turns"),
+        (data.get("run_config") or {}).get("turn_reminder"),
     )