Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 86 additions & 0 deletions skillscope/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import sys
import tempfile
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path

from . import datasets, deadline
Expand Down Expand Up @@ -94,6 +95,88 @@ def claude_env() -> dict[str, str]:
return env


# What `--model opus` actually resolved to, learned from the preflight call.
# Ambient like the active deadline: every engine wants it in its report and
# none of them should pay for a second call to find it out.
_RESOLVED_MODEL: str | None = None


@lru_cache(maxsize=1)
def cli_version() -> str | None:
"""The `claude` CLI's own version, or None if it cannot be read.

Part of a run's provenance. The CLI is what discovers and activates a
skill, so a routing score is only comparable to another taken on the same
build. `claude --version` prints `2.1.270 (Claude Code)`; the number is
the part worth recording.
"""
claude_bin = shutil.which("claude")
if not claude_bin:
return None
try:
proc = subprocess.run(
[claude_bin, "--version"], capture_output=True, text=True,
encoding="utf-8", timeout=30, env=claude_env(),
)
except (OSError, subprocess.SubprocessError):
return None
fields = (proc.stdout or "").split()
return fields[0] if proc.returncode == 0 and fields else None


def resolved_model() -> str | None:
"""The model the API actually served, or None if nothing has asked it yet.

`opus` and `sonnet` are aliases whose target moves, so the alias a caller
passed does not identify the thing that produced a score. This is None
when the preflight was skipped, which is honest: nothing has spoken to the
API, so nothing knows.
"""
return _RESOLVED_MODEL


def _model_from_result(stdout: str, alias: str | None) -> str | None:
"""Which model answered, read out of a `--output-format json` result.

The result reports usage per model rather than naming the one that ran,
and the CLI bills auxiliary work to a different model in the same block: a
preflight that asked for `sonnet` can come back with a session-title model
listed beside it. The alias is what disambiguates, since `sonnet` resolves
to a name containing `sonnet`. With no alias to match, the entry that did
the most work is the best available answer.
"""
try:
payload = json.loads(stdout)
except (TypeError, ValueError):
return None
usage = payload.get("modelUsage")
if not isinstance(usage, dict):
return None

# `canonicalModel` is the name the provider settled on; the key is the one
# the CLI asked under. Prefer the former, fall back to the latter.
served: dict[str, int] = {}
for name, detail in usage.items():
if not isinstance(name, str) or not name:
continue
tokens = 0
if isinstance(detail, dict):
if isinstance(detail.get("canonicalModel"), str) and detail["canonicalModel"]:
name = detail["canonicalModel"]
if isinstance(detail.get("outputTokens"), int):
tokens = detail["outputTokens"]
served[name] = max(served.get(name, 0), tokens)

if not served:
return None
if alias:
needle = alias.lower()
for name in served:
if needle in name.lower():
return name
return max(served, key=lambda name: served[name])


def check_api_reachable(model: str | None = DEFAULT_MODEL, timeout: float = 60) -> tuple[bool, str]:
"""Preflight: confirm the `claude` CLI can actually reach the API.

Expand Down Expand Up @@ -129,6 +212,9 @@ def check_api_reachable(model: str | None = DEFAULT_MODEL, timeout: float = 60)
if proc.returncode != 0:
detail = (proc.stderr or proc.stdout or f"exit code {proc.returncode}").strip()
return False, detail[:500]

global _RESOLVED_MODEL
_RESOLVED_MODEL = _model_from_result(proc.stdout or "", model)
return True, "ok"


Expand Down
26 changes: 25 additions & 1 deletion skillscope/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from . import behavior, config, datasets, deadline, references, routing, structure
from . import agent, behavior, config, datasets, deadline, references, routing, structure
from . import selection as select_module
from .agent import check_api_reachable, enforce_model_policy

Expand All @@ -81,6 +81,28 @@
RUNS_DIRNAME = Path(".skillscope") / "runs"


# The only engine today, named rather than implied. A reader should not have to
# date a report to work out what graded it.
ENGINE = "legacy"


def _provenance() -> dict:
"""What produced a run's numbers, as opposed to what those numbers say.

Two scores are comparable only if the same engine, the same agent build
and the same model produced both. `model` in the meta block is the alias
the caller asked for, which is not enough on its own: the target moves
underneath it, as does the CLI that does the discovering. A report that
records neither cannot tell a skill that got worse from a dependency that
changed.
"""
return {
"engine": ENGINE,
"agent_cli_version": agent.cli_version(),
"model_resolved": agent.resolved_model(),
}


def _selected_skills(names: str) -> list[str]:
available = datasets.skills_with_datasets()
if not names:
Expand Down Expand Up @@ -397,6 +419,7 @@ def cmd_routing(args: argparse.Namespace) -> int:
list(routing_set),
{
"model": args.model,
**_provenance(),
"effort": args.effort,
"skills": list(routing_set),
"extended": args.extended,
Expand Down Expand Up @@ -447,6 +470,7 @@ def cmd_behavioral(args: argparse.Namespace) -> int:
outcomes,
{
"model": args.model,
**_provenance(),
"effort": args.effort,
"skills": skills,
"extended": args.extended,
Expand Down
114 changes: 114 additions & 0 deletions tests/test_skillscope.py
Original file line number Diff line number Diff line change
Expand Up @@ -2501,5 +2501,119 @@ def test_an_empty_room_leaves_only_the_shared_pool(self) -> None:
self.assertTrue(all(case.skill is None for case in cases))


class TestARunRecordsWhatProducedIt(unittest.TestCase):
"""A score is only comparable to another if the same things produced both.

`model` in a report is the alias the caller asked for, and what `opus`
points at moves. The CLI that discovers and activates a skill updates on
its own schedule. Without those written down, two runs that disagree hold
nothing that separates a skill getting worse from a dependency that
changed, so these fields are the report's own provenance and the tests
here are about them being right rather than merely present.
"""

def setUp(self) -> None:
self.addCleanup(agent.cli_version.cache_clear)
agent.cli_version.cache_clear()
# The resolved model is module state, set by whichever preflight ran
# last. Put back whatever the process already had.
previous = agent._RESOLVED_MODEL
self.addCleanup(setattr, agent, "_RESOLVED_MODEL", previous)

@staticmethod
def usage(*models: tuple[str, int]) -> str:
return json.dumps(
{"modelUsage": {name: {"outputTokens": out} for name, out in models}}
)

def test_the_alias_picks_its_own_entry_out_of_a_shared_usage_block(self) -> None:
# The CLI bills auxiliary work to a model the run never asked for: a
# preflight for `sonnet` comes back with a session-title model listed
# beside it. Taking the first entry reports that one.
blob = self.usage(("gpt-5.5", 9), ("claude-sonnet-5", 4))
self.assertEqual(agent._model_from_result(blob, "sonnet"), "claude-sonnet-5")

def test_the_canonical_name_wins_over_the_key_it_was_billed_under(self) -> None:
blob = json.dumps(
{
"modelUsage": {
"opus": {"canonicalModel": "claude-opus-5-20260722", "outputTokens": 4}
}
}
)
self.assertEqual(
agent._model_from_result(blob, "opus"), "claude-opus-5-20260722"
)

def test_with_no_alias_the_entry_that_did_the_work_is_the_answer(self) -> None:
blob = self.usage(("a-title-model", 2), ("the-one-that-answered", 400))
self.assertEqual(
agent._model_from_result(blob, None), "the-one-that-answered"
)

def test_output_that_says_nothing_about_a_model_resolves_to_nothing(self) -> None:
for stdout in ("", "not json at all", "{}", '{"modelUsage": []}',
'{"modelUsage": {}}'):
with self.subTest(stdout=stdout):
self.assertIsNone(agent._model_from_result(stdout, "opus"))

def test_a_reachable_api_leaves_behind_the_model_it_served(self) -> None:
served = self.usage(("claude-opus-5-20260722", 4))
with mock.patch.object(agent.shutil, "which", return_value="/usr/bin/claude"), \
mock.patch.object(
agent.subprocess,
"run",
return_value=subprocess.CompletedProcess([], 0, served, ""),
):
agent._RESOLVED_MODEL = None
self.assertEqual(agent.check_api_reachable("opus"), (True, "ok"))
self.assertEqual(agent.resolved_model(), "claude-opus-5-20260722")

def test_a_preflight_nobody_ran_reports_no_model_rather_than_a_guess(self) -> None:
# `--skip-preflight` means nothing has spoken to the API. Reporting the
# alias here would state as fact something no call confirmed.
agent._RESOLVED_MODEL = None
self.assertIsNone(agent.resolved_model())

def test_the_cli_version_is_the_number_not_the_product_name(self) -> None:
with mock.patch.object(agent.shutil, "which", return_value="/usr/bin/claude"), \
mock.patch.object(
agent.subprocess,
"run",
return_value=subprocess.CompletedProcess(
[], 0, "2.1.270 (Claude Code)\n", ""
),
):
self.assertEqual(agent.cli_version(), "2.1.270")

def test_no_cli_on_the_path_is_reported_as_unknown_not_as_a_crash(self) -> None:
# Structural runs never touch an agent, so an absent CLI must not be
# what stops a report being written.
with mock.patch.object(agent.shutil, "which", return_value=None):
self.assertIsNone(agent.cli_version())

def test_a_cli_that_fails_to_answer_is_also_unknown(self) -> None:
with mock.patch.object(agent.shutil, "which", return_value="/usr/bin/claude"), \
mock.patch.object(
agent.subprocess, "run", side_effect=OSError("boom")
):
self.assertIsNone(agent.cli_version())

def test_every_report_carries_the_three_provenance_fields(self) -> None:
with mock.patch.object(agent, "cli_version", return_value="2.1.270"), \
mock.patch.object(agent, "resolved_model", return_value="claude-opus-5"):
self.assertEqual(
cli._provenance(),
{
"engine": cli.ENGINE,
"agent_cli_version": "2.1.270",
"model_resolved": "claude-opus-5",
},
)

def test_the_engine_is_named_so_a_reader_never_has_to_date_the_report(self) -> None:
self.assertEqual(cli.ENGINE, "legacy")


if __name__ == "__main__":
unittest.main(verbosity=2)
Loading