feat(eval): skill-adherence measurement mode

Add a two-arm skill-adherence mode to the eval measurement substrate that
measures whether a NAMED skill changes tool-use behaviour, so skill-in-system
(main) can be compared against skill-in-context.

- _run_single_test gains skill/skill_mode: skill_mode builds HeadlessSession
  under natural composition (no system_prompt_override) and, for the treatment
  arm, seeds the skill into the temp DB and activates it via the real
  set_skill path so the skill body folds into the system message under test.
  skill_mode defaults False, so the optimizer/measure paths are unchanged.
- Thread skill/skill_mode through _run_and_score_subprocess, _run_iteration
  and _run_iteration_parallel (serial + parallel).
- run_skill_adherence: per case, run treatment (skill) vs control (no skill)
  n_runs each, score against expected_actions, report per-case lift =
  pass_rate(treatment) - pass_rate(control) and the mean lift. The control
  isolates the skill's causal effect.
- turnstone-eval --skill-adherence <dataset>: loads a skill-scenario dataset
  and prints a treatment/control/lift table.
- eval_skill_adherence.json: authored search-first / test-after-edit /
  changelog-update scenarios, chosen so the base model does not do the action
  by default.
- tests: plumbing proof (skill folds into system_messages for treatment,
  absent for control) + lift-math aggregation.
This commit is contained in:
Patrick Buckley
2026-07-03 17:16:22 -07:00
parent d47d528d9a
commit 845df69031
4 changed files with 502 additions and 2 deletions
+56
View File
@@ -0,0 +1,56 @@
{
"defaults": {
"n_runs": 3
},
"cases": [
{
"id": "search-first",
"skill": {
"name": "search-first",
"content": "# Search First\n\nBefore answering ANY question about where something lives in the codebase, you MUST call the `search` tool first. Never answer from memory."
},
"user_prompt": "Where is JWT token validation implemented in this project?",
"expected_actions": [{ "tool": "search" }],
"match_mode": "ordered_subset",
"max_turns": 4
},
{
"id": "test-after-edit",
"skill": {
"name": "test-after-edit",
"content": "# Test After Edit\n\nAfter editing or writing ANY file, you MUST run the test suite with `python -m pytest` via bash before you finish. Do not report done until tests have run."
},
"user_prompt": "Add a function `clamp(x, lo, hi)` that clamps x to [lo, hi] in utils.py.",
"setup": {
"files": {
"utils.py": ""
}
},
"expected_actions": [
{ "tool": "write_file" },
{ "tool": "bash", "args_pattern": { "command": "pytest" } }
],
"match_mode": "ordered_subset",
"max_turns": 8
},
{
"id": "changelog-update",
"skill": {
"name": "changelog-update",
"content": "# Changelog Discipline\n\nWhenever you modify a file, you MUST also append a one-line entry to CHANGELOG.md describing the change in the same task."
},
"user_prompt": "Fix the off-by-one so pager.py shows the last page. Edit pager.py.",
"setup": {
"files": {
"pager.py": "def last_page(total_items, per_page):\n # off-by-one: drops the final partial page\n return total_items // per_page\n",
"CHANGELOG.md": "# Changelog\n"
}
},
"expected_actions": [
{ "tool": "edit_file", "args_pattern": { "path": "CHANGELOG.md" } }
],
"match_mode": "subset",
"max_turns": 8
}
]
}
+200
View File
@@ -0,0 +1,200 @@
"""Tests for turnstone.eval skill-adherence measurement mode.
Two levels, neither requires a live model:
* ``TestSkillComposition`` is the load-bearing plumbing proof — it seeds a
named skill, builds ``HeadlessSession`` under natural composition, and
asserts the skill body folds into ``system_messages`` for the treatment
arm and is absent for the control arm. This is what makes the two arms
measure different things.
* ``TestAdherenceLift`` unit-tests ``run_skill_adherence``'s lift math with
the per-arm runner stubbed out.
"""
import os
import tempfile
from collections.abc import Iterator
from typing import Any
import pytest
from openai import OpenAI
from turnstone.core.storage import get_storage, init_storage, reset_storage
from turnstone.eval import core
from turnstone.eval.core import HeadlessSession, run_skill_adherence
_SKILL = {
"name": "search-first",
"content": (
"# Search First\n\nBefore answering ANY question about where something "
"lives in the codebase, you MUST call the `search` tool first. "
"SENTINEL_SKILL_BODY_MARKER."
),
}
@pytest.fixture
def temp_storage() -> Iterator[None]:
"""Fresh sqlite storage in a temp dir, torn down after the test."""
workdir = tempfile.mkdtemp(prefix="turnstone_skill_test_")
reset_storage()
init_storage("sqlite", path=os.path.join(workdir, ".eval.db"), run_migrations=False)
try:
yield
finally:
reset_storage()
import shutil
shutil.rmtree(workdir, ignore_errors=True)
def _seed_skill(skill: dict[str, str]) -> None:
"""Seed a named skill exactly as the runner does."""
get_storage().create_prompt_template(
template_id="eval-skill",
name=skill["name"],
category="eval",
content=skill["content"],
variables="[]",
is_default=False,
org_id="",
created_by="eval",
activation="named",
enabled=True,
)
def _system_text(session: HeadlessSession) -> str:
return "\n".join(m["content"] for m in session.system_messages)
class TestSkillComposition:
"""Prove the treatment/control arms compose different system messages."""
def test_treatment_folds_skill_into_system(self, temp_storage: None) -> None:
_seed_skill(_SKILL)
client = OpenAI(base_url="http://localhost:9/v1", api_key="dummy")
session = HeadlessSession(client=client, model="test-model")
try:
# Treatment arm activates the seeded skill via the real path.
session.set_skill(_SKILL["name"])
assert "SENTINEL_SKILL_BODY_MARKER" in _system_text(session)
finally:
session.close()
def test_control_omits_skill(self, temp_storage: None) -> None:
# Control arm: no skill seeded, no set_skill — natural default only.
client = OpenAI(base_url="http://localhost:9/v1", api_key="dummy")
session = HeadlessSession(client=client, model="test-model")
try:
assert "SENTINEL_SKILL_BODY_MARKER" not in _system_text(session)
finally:
session.close()
def test_no_system_prompt_override_in_skill_mode(self, temp_storage: None) -> None:
# skill_mode must NOT override the base identity — a real base prompt
# (persona / composed developer message) must survive, or we'd be
# measuring an empty prompt instead of the identity under test.
client = OpenAI(base_url="http://localhost:9/v1", api_key="dummy")
session = HeadlessSession(client=client, model="test-model")
try:
assert _system_text(session).strip(), "expected a composed base prompt"
finally:
session.close()
class TestAdherenceLift:
"""Unit-test the lift math with the per-arm runner stubbed."""
def test_lift_treatment_over_control(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Stub _run_iteration: treatment (skill != None) passes 3/3, control
# (skill is None) passes 1/3. run_skill_adherence must report the
# difference as the lift.
def fake_run_iteration(**kwargs: Any) -> dict[str, Any]:
rate = 1.0 if kwargs.get("skill") is not None else 1.0 / 3.0
return {"aggregate": {"overall_pass_rate": rate}}
monkeypatch.setattr(core, "_run_iteration", fake_run_iteration)
cases = [
{
"id": "search-first",
"skill": _SKILL,
"user_prompt": "where is X?",
"expected_actions": [{"tool": "search"}],
}
]
result = run_skill_adherence(
client=None,
base_url="http://localhost:9/v1",
api_key="dummy",
model="test-model",
cases=cases,
n_runs=3,
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=8192,
)
assert len(result["cases"]) == 1
row = result["cases"][0]
assert row["case_id"] == "search-first"
assert row["skill"] == "search-first"
assert row["treatment_rate"] == pytest.approx(1.0)
assert row["control_rate"] == pytest.approx(1.0 / 3.0)
assert row["lift"] == pytest.approx(2.0 / 3.0)
assert row["n_runs"] == 3
assert result["mean_lift"] == pytest.approx(2.0 / 3.0)
def test_skipped_when_no_skill(self, monkeypatch: pytest.MonkeyPatch) -> None:
# A case with no skill is not measurable — it must be skipped, not
# crash, and must not contribute to the mean.
def fake_run_iteration(**kwargs: Any) -> dict[str, Any]:
return {"aggregate": {"overall_pass_rate": 1.0}}
monkeypatch.setattr(core, "_run_iteration", fake_run_iteration)
cases = [{"id": "no-skill", "user_prompt": "hi", "expected_actions": []}]
result = run_skill_adherence(
client=None,
base_url="http://localhost:9/v1",
api_key="dummy",
model="test-model",
cases=cases,
n_runs=3,
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=8192,
)
assert result["cases"] == []
assert result["mean_lift"] == 0.0
def test_mean_lift_averages_multiple_cases(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Two skill cases with different lifts average into mean_lift.
rates = iter([1.0, 0.0, 1.0, 0.5]) # t1, c1, t2, c2 -> lifts 1.0, 0.5
def fake_run_iteration(**kwargs: Any) -> dict[str, Any]:
return {"aggregate": {"overall_pass_rate": next(rates)}}
monkeypatch.setattr(core, "_run_iteration", fake_run_iteration)
cases = [
{"id": "a", "skill": _SKILL, "user_prompt": "q", "expected_actions": []},
{"id": "b", "skill": _SKILL, "user_prompt": "q", "expected_actions": []},
]
result = run_skill_adherence(
client=None,
base_url="http://localhost:9/v1",
api_key="dummy",
model="test-model",
cases=cases,
n_runs=2,
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=8192,
)
assert [c["lift"] for c in result["cases"]] == pytest.approx([1.0, 0.5])
assert result["mean_lift"] == pytest.approx(0.75)
+73
View File
@@ -25,11 +25,69 @@ from turnstone.core.session import ChatSession
from turnstone.eval.core import ( from turnstone.eval.core import (
NullUI, NullUI,
_append_summary_tsv, _append_summary_tsv,
_print_skill_adherence_table,
_print_summary_table, _print_summary_table,
_run_iteration, _run_iteration,
run_skill_adherence,
) )
def _run_skill_adherence_cli(
args: argparse.Namespace,
client: OpenAI,
model: str,
api_key: str,
) -> None:
"""Load a skill-scenario dataset and report per-case adherence lift."""
with open(args.test_file) as f:
suite: dict[str, Any] = json.load(f)
cases: list[dict[str, Any]] = suite["cases"]
for i, case in enumerate(cases):
if "id" not in case:
raise SystemExit(f"Test case {i} missing required 'id' field")
if "user_prompt" not in case:
raise SystemExit(f"Test case '{case.get('id', i)}' missing 'user_prompt'")
if not any(c.get("skill") for c in cases):
raise SystemExit("No cases carry a 'skill' — nothing to measure for adherence")
defaults = suite.get("defaults", {})
resolved_n_runs: int = (
args.n_runs if args.n_runs is not None else int(defaults.get("n_runs", 3))
)
parallel = args.parallel if args.parallel != 0 else (os.cpu_count() or 4)
result = run_skill_adherence(
client=client,
base_url=args.base_url,
api_key=api_key,
model=model,
cases=cases,
n_runs=resolved_n_runs,
temperature=args.temperature,
max_tokens=args.max_tokens,
reasoning_effort=args.reasoning_effort,
context_window=args.context_window,
test_timeout=args.test_timeout,
parallel=parallel,
verbose=args.verbose,
)
result["meta"] = {
"model": model,
"base_url": args.base_url,
"test_suite": args.test_file,
"n_runs": resolved_n_runs,
"started": datetime.now().isoformat(),
}
_print_skill_adherence_table(result)
with open(args.output, "w") as f:
json.dump(result, f, indent=2)
f.write("\n")
print(f"Results written to {args.output}")
def main() -> None: def main() -> None:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="Headless measurement for turnstone (scores tool use against expected actions)", description="Headless measurement for turnstone (scores tool use against expected actions)",
@@ -116,6 +174,15 @@ def main() -> None:
default=1, default=1,
help="Parallel workers (default: 1=serial, 0=auto)", help="Parallel workers (default: 1=serial, 0=auto)",
) )
parser.add_argument(
"--skill-adherence",
action="store_true",
help=(
"Measure skill adherence: for each case carrying a 'skill', run a "
"treatment arm (skill composed into the system message) vs a control "
"arm (no skill) and report the pass-rate lift"
),
)
parser.add_argument( parser.add_argument(
"-v", "-v",
"--verbose", "--verbose",
@@ -142,6 +209,12 @@ def main() -> None:
assert detected is not None # fatal=True guarantees non-None or SystemExit assert detected is not None # fatal=True guarantees non-None or SystemExit
model = detected model = detected
# Skill-adherence mode is a distinct two-arm measurement — it uses natural
# prompt composition (no --prompt), so branch before the initial-prompt path.
if args.skill_adherence:
_run_skill_adherence_cli(args, client, model, api_key)
return
# Resolve the initial prompt: --prompt file, else turnstone's built-in. # Resolve the initial prompt: --prompt file, else turnstone's built-in.
initial_prompt: str | None = None initial_prompt: str | None = None
if args.prompt: if args.prompt:
+173 -2
View File
@@ -32,7 +32,7 @@ from openai import OpenAI
from turnstone.core.providers import LLMProvider, create_client, create_provider from turnstone.core.providers import LLMProvider, create_client, create_provider
from turnstone.core.session import ChatSession from turnstone.core.session import ChatSession
from turnstone.core.storage import init_storage, reset_storage from turnstone.core.storage import get_storage, init_storage, reset_storage
from turnstone.core.tools import INTERACTIVE_TOOLS, PRIMARY_KEY_MAP from turnstone.core.tools import INTERACTIVE_TOOLS, PRIMARY_KEY_MAP
from turnstone.core.trajectory import Role, turn_from_dict from turnstone.core.trajectory import Role, turn_from_dict
@@ -441,12 +441,23 @@ def _run_single_test(
log_prefix: str = "", log_prefix: str = "",
test_timeout: int = 300, test_timeout: int = 300,
tool_overrides: dict[str, dict[str, Any]] | None = None, tool_overrides: dict[str, dict[str, Any]] | None = None,
skill: dict[str, Any] | None = None,
skill_mode: bool = False,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Run a single test case once in an isolated temp directory. """Run a single test case once in an isolated temp directory.
Uses os.chdir (process-global), so concurrent calls must run in Uses os.chdir (process-global), so concurrent calls must run in
separate processes (see _run_and_score_subprocess / --parallel). separate processes (see _run_and_score_subprocess / --parallel).
When ``skill_mode`` is True the session is built WITHOUT a system-prompt
override so the model runs under turnstone's natural prompt composition
(the base identity under test). If ``skill`` is given it is seeded into
the temp DB and activated via the real ``set_skill`` path, so the skill
body composes into the system message exactly as it would in production;
``skill`` None is the control arm (natural default, no skill). When
``skill_mode`` is False behaviour is unchanged — the system prompt is
overridden as before.
Returns dict with keys: tool_log, final_content, message_count, Returns dict with keys: tool_log, final_content, message_count,
elapsed, usage. elapsed, usage.
""" """
@@ -471,6 +482,24 @@ def _run_single_test(
os.chdir(workdir) os.chdir(workdir)
# Skill-adherence treatment arm: seed the named skill into the temp DB
# (once — before the retry loop) so set_skill can activate it through
# the real composition path. The subprocess/serial DB is fresh per
# run, so template_id "eval-skill" never collides.
if skill_mode and skill is not None:
get_storage().create_prompt_template(
template_id="eval-skill",
name=skill["name"],
category="eval",
content=skill["content"],
variables="[]",
is_default=False,
org_id="",
created_by="eval",
activation="named",
enabled=True,
)
max_turns = case.get("max_turns", 15) max_turns = case.get("max_turns", 15)
# Retry on transient API errors to avoid poisoning eval scores # Retry on transient API errors to avoid poisoning eval scores
tool_log: list[dict[str, Any]] = [] tool_log: list[dict[str, Any]] = []
@@ -489,7 +518,9 @@ def _run_single_test(
session = HeadlessSession( session = HeadlessSession(
client=run_client, client=run_client,
model=model, model=model,
system_prompt_override=system_prompt, # skill_mode uses turnstone's natural composition (no override)
# so the skill can fold into the system message under test.
system_prompt_override=None if skill_mode else system_prompt,
instructions=None, instructions=None,
temperature=temperature, temperature=temperature,
max_tokens=max_tokens, max_tokens=max_tokens,
@@ -499,6 +530,11 @@ def _run_single_test(
tool_truncation=2000, tool_truncation=2000,
tool_overrides=tool_overrides, tool_overrides=tool_overrides,
) )
if skill_mode and skill is not None:
# Activate the seeded skill via the production path:
# _load_skills() -> _init_system_messages() composes the
# skill body into session.system_messages.
session.set_skill(skill["name"])
executor: ThreadPoolExecutor | None = None executor: ThreadPoolExecutor | None = None
try: try:
executor = ThreadPoolExecutor(max_workers=1) executor = ThreadPoolExecutor(max_workers=1)
@@ -587,6 +623,8 @@ def _run_and_score_subprocess(params: dict[str, Any]) -> dict[str, Any]:
log_prefix="", log_prefix="",
test_timeout=params["test_timeout"], test_timeout=params["test_timeout"],
tool_overrides=params.get("tool_overrides"), tool_overrides=params.get("tool_overrides"),
skill=params.get("skill"),
skill_mode=params.get("skill_mode", False),
) )
score_result = score_run( score_result = score_run(
@@ -842,6 +880,8 @@ def _run_iteration_parallel(
parallel: int, parallel: int,
prompt_variants: dict[str, list[str]] | None = None, prompt_variants: dict[str, list[str]] | None = None,
tool_overrides: dict[str, dict[str, Any]] | None = None, tool_overrides: dict[str, dict[str, Any]] | None = None,
skill: dict[str, Any] | None = None,
skill_mode: bool = False,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Run all test cases in parallel using ProcessPoolExecutor.""" """Run all test cases in parallel using ProcessPoolExecutor."""
# Build work items for every (case, run) combination # Build work items for every (case, run) combination
@@ -872,6 +912,8 @@ def _run_iteration_parallel(
"test_timeout": test_timeout, "test_timeout": test_timeout,
"original_user_prompt": case["user_prompt"], "original_user_prompt": case["user_prompt"],
"tool_overrides": tool_overrides, "tool_overrides": tool_overrides,
"skill": skill,
"skill_mode": skill_mode,
} }
) )
@@ -993,6 +1035,8 @@ def _run_iteration(
api_key: str = "", api_key: str = "",
prompt_variants: dict[str, list[str]] | None = None, prompt_variants: dict[str, list[str]] | None = None,
tool_overrides: dict[str, dict[str, Any]] | None = None, tool_overrides: dict[str, dict[str, Any]] | None = None,
skill: dict[str, Any] | None = None,
skill_mode: bool = False,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Run all test cases n_runs times and score them.""" """Run all test cases n_runs times and score them."""
if parallel > 1 and base_url: if parallel > 1 and base_url:
@@ -1011,6 +1055,8 @@ def _run_iteration(
parallel=parallel, parallel=parallel,
prompt_variants=prompt_variants, prompt_variants=prompt_variants,
tool_overrides=tool_overrides, tool_overrides=tool_overrides,
skill=skill,
skill_mode=skill_mode,
) )
case_results: dict[str, Any] = {} case_results: dict[str, Any] = {}
@@ -1061,6 +1107,8 @@ def _run_iteration(
log_prefix=log_prefix, log_prefix=log_prefix,
test_timeout=test_timeout, test_timeout=test_timeout,
tool_overrides=tool_overrides, tool_overrides=tool_overrides,
skill=skill,
skill_mode=skill_mode,
) )
score_result = score_run( score_result = score_run(
@@ -1183,6 +1231,98 @@ def _run_iteration(
return _aggregate_case_results(cases, case_results, total_tokens) return _aggregate_case_results(cases, case_results, total_tokens)
# ─── Skill-adherence driver ──────────────────────────────────────────────────
def run_skill_adherence(
client: Any,
base_url: str,
api_key: str,
model: str,
cases: list[dict[str, Any]],
n_runs: int,
temperature: float,
max_tokens: int,
reasoning_effort: str,
context_window: int,
test_timeout: int = 300,
parallel: int = 1,
verbose: bool = False,
) -> dict[str, Any]:
"""Measure how much a named skill changes tool-use behaviour.
For every case that carries a ``skill`` this runs two arms ``n_runs``
times each, scoring both against the case's ``expected_actions``:
* **treatment** — the skill is composed into the system message via the
real ``set_skill`` path (``skill_mode=True, skill=<case skill>``);
* **control** — the same base identity with no skill
(``skill_mode=True, skill=None``).
The adherence lift is ``pass_rate(treatment) - pass_rate(control)``. The
control isolates the skill's causal effect: a scenario the model passes
anyway yields ~0 lift and is uninformative — that near-zero IS the signal.
Returns ``{"cases": [{case_id, skill, treatment_rate, control_rate, lift,
n_runs}, ...], "mean_lift": float}``.
"""
case_results: list[dict[str, Any]] = []
skill_cases = [c for c in cases if c.get("skill")]
for ci, case in enumerate(skill_cases):
skill = case["skill"]
# Drop the skill key from the case handed to the runner — it is
# supplied out-of-band per arm, not read from the case dict.
arm_case = {k: v for k, v in case.items() if k != "skill"}
print(
f"\n {CYAN}[{ci + 1}/{len(skill_cases)}]{RESET} "
f"{BOLD}{case['id']}{RESET} — skill {DIM}'{skill['name']}'{RESET}"
)
arm_rates: dict[str, float] = {}
for arm, arm_skill in (("treatment", skill), ("control", None)):
print(f" {DIM}{arm}{RESET}")
iter_result = _run_iteration(
client=client,
model=model,
system_prompt="",
cases=[arm_case],
n_runs=n_runs,
temperature=temperature,
max_tokens=max_tokens,
reasoning_effort=reasoning_effort,
context_window=context_window,
verbose=verbose,
test_timeout=test_timeout,
# Never fast-fail: the treatment arm's pass rate must be
# counted over every run, and control runs are expected to
# fail — skipping them would corrupt the lift.
fast_fail=False,
parallel=parallel,
base_url=base_url,
api_key=api_key,
skill=arm_skill,
skill_mode=True,
)
arm_rates[arm] = iter_result["aggregate"]["overall_pass_rate"]
treatment_rate = arm_rates["treatment"]
control_rate = arm_rates["control"]
case_results.append(
{
"case_id": case["id"],
"skill": skill["name"],
"treatment_rate": treatment_rate,
"control_rate": control_rate,
"lift": treatment_rate - control_rate,
"n_runs": n_runs,
}
)
mean_lift = sum(c["lift"] for c in case_results) / len(case_results) if case_results else 0.0
return {"cases": case_results, "mean_lift": mean_lift}
# ─── Summary & reporting ───────────────────────────────────────────────────── # ─── Summary & reporting ─────────────────────────────────────────────────────
@@ -1243,6 +1383,37 @@ def _print_summary_table(iter_result: dict[str, Any]) -> None:
) )
def _print_skill_adherence_table(result: dict[str, Any]) -> None:
"""Print a per-case treatment/control/lift table plus the mean lift."""
rows = result.get("cases", [])
if not rows:
print("\n No skill-bearing cases to measure.")
return
max_id = max(len(str(r["case_id"])) for r in rows)
max_id = max(max_id, 4) # min "CASE" header
print(f"\n{BOLD} {'CASE'.ljust(max_id)} {'TREAT':>6} {'CTRL':>6} {'LIFT':>7}{RESET}")
print(f" {'' * (max_id + 25)}")
for r in rows:
lift = r["lift"]
color = GREEN if lift > 0.01 else (RED if lift < -0.01 else DIM)
print(
f" {str(r['case_id']).ljust(max_id)} "
f"{r['treatment_rate']:>6.2f} {r['control_rate']:>6.2f} "
f"{color}{lift:>+7.2f}{RESET}"
)
print(f" {'' * (max_id + 25)}")
mean = result.get("mean_lift", 0.0)
mcolor = GREEN if mean > 0.01 else (RED if mean < -0.01 else DIM)
print(
f" {BOLD}{'MEAN'.ljust(max_id)}{RESET} {'':>6} {'':>6} "
f"{mcolor}{BOLD}{mean:>+7.2f}{RESET}"
)
def _append_summary_tsv( def _append_summary_tsv(
path: str, path: str,
iter_result: dict[str, Any], iter_result: dict[str, Any],