ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity
Files
6eae1c39543efcb7fbcc87db6475c24c889dce18
turnstone/tests/test_eval_nudges.py
T
Patrick Buckley 6eae1c3954 fix(providers): support OpenAI v3 HTTPX2 transport
2026-08-11 23:30:44 -07:00

3064 lines
134 KiB
Python
Raw Blame History

"""Harness tests for the idle-nudge behavioral eval (no LLM calls).
The LLM loop itself is exercised by real sweeps against a live
endpoint; these pin everything deterministic around it — cell fixture
validity, stimulus shape, scoring semantics, and the stub client's
network inertness — so a sweep failure means the model, not the
harness.
"""
from __future__ import annotations
import hashlib
import json
import os
import shutil
import tempfile
import time
from typing import Any
import pytest
from turnstone.console.coordinator_idle_observer import _ACTIVE_CHILD_STATES
from turnstone.core.metacognition import (
NUDGE_CHILD_RUNNING_LINE,
NUDGE_CHILD_STOPPED_LINE,
NUDGE_IDLE_TASKS_CHILD_DOOR,
NUDGE_IDLE_TASKS_CHILD_SLOT,
NUDGE_IDLE_TASKS_ID_SLOT,
)
from turnstone.core.session import COORDINATOR_TOOLS
from turnstone.core.storage._registry import (
get_storage,
init_storage,
is_storage_initialized,
reset_storage,
)
from turnstone.core.workstream import WorkstreamKind
from turnstone.eval.nudges import (
_LIVE_CHILD_STATES,
_MUTATING_TASKS_ACTIONS,
_NO_CAVEAT_SKIP_REASON,
_TASKS_ACTION_KEY,
_TASKS_SCHEMA_ACTIONS,
ARM_BARE_CONTINUE,
ARM_NO_CAVEAT,
ARM_NUDGE,
ARM_PAIR_TF,
KNOWN_ARMS,
_live_children,
_seed_child_transcripts,
_seed_tasks,
_seed_transcript,
_seed_world,
_StubCoordinatorClient,
_validate_cells,
build_stimulus,
render_tasks_body,
run_nudge_response,
score_nudge_run,
)
from turnstone.eval.scenarios.nudges import NUDGE_CELLS
_LIVE_TOOL_NAMES = {t["function"]["name"] for t in COORDINATOR_TOOLS}
# The minimum a cell must seed to render a non-empty ``idle_tasks``
# body. Every validator fixture below that is not ABOUT the seed list
# carries it, so each refusal test trips the one refusal it names.
_OPEN_TASK: dict[str, Any] = {"title": "audit auth.py for CSRF handling", "status": "pending"}
# A child row in a LIVE state — what the ``no_caveat`` arm needs to be
# measuring anything at all. ``running`` rather than ``idle`` only
# because it also satisfies the pair arms' narrower active predicate,
# so one constant serves both classes of fixture. It carries its
# assignment because every child must — the hollow-child refusal binds
# any cell that seeds children, scenery included.
_LIVE_CHILD: dict[str, Any] = {
"ws_id": "ws-c1",
"name": "auditor",
"state": "running",
"transcript": [{"role": "user", "content": "audit auth.py for CSRF handling"}],
}
# The transcript rows that make a child row legal in each state class:
# every child carries its assignment; an idle child also carries the
# completion turn the wait synthesis will surface.
_ASSIGNMENT_ROW: dict[str, str] = {"role": "user", "content": "audit auth.py for CSRF handling"}
_FINDINGS_ROW: dict[str, str] = {"role": "assistant", "content": "Audit complete: 3 findings."}
def _trip_cells() -> list[dict[str, Any]]:
"""One cell per registered refusal, in registration order, each
tripping ONLY the check it is paired with.
Shared by the structural guard (every check is reachable) and the
ordering guard (every refusal fires before the canary probe), so
the two cannot drift apart or cover different sets.
"""
return [
{"id": "X_t", "arms": ARM_NUDGE, "tasks": [_OPEN_TASK]},
{"id": "X_t", "arms": [], "tasks": [_OPEN_TASK]},
{"id": "X_t", "arms": ["not_an_arm"], "tasks": [_OPEN_TASK]},
{"id": "X_t", "arms": [ARM_NUDGE, ARM_NUDGE], "tasks": [_OPEN_TASK]},
{"id": "X_t", "arms": [ARM_NUDGE], "children": None, "tasks": [_OPEN_TASK]},
{"id": "X_t", "arms": [ARM_NUDGE], "tasks": [{"status": "pending"}]},
{
"id": "X_t",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"transcript": [{"content": "no role"}],
},
{
"id": "X_t",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"tool_stubs": ["not-a-mapping"],
},
{
"id": "X_t",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"expect_state": {3: {"status": "done"}},
},
{
"id": "X_t",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"forbid_actions": [{"args": {}}],
},
{"id": "X_t", "arms": [ARM_NUDGE], "tasks": [{"title": "t", "status": "done"}]},
# Parked + open together: production parks the nudge on any
# ``needs_user`` row, so this cell would score a body no
# coordinator can receive. Open task present (so the
# open-task check passes) and no pair/caveat arm, so only the
# park check can refuse it.
{
"id": "X_t",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK, {"title": "waiting on you", "status": "needs_user"}],
},
{"id": "X_t", "arms": [ARM_PAIR_TF], "children": [], "tasks": [_OPEN_TASK]},
{"id": "X_t", "arms": [ARM_NO_CAVEAT], "children": [], "tasks": [_OPEN_TASK]},
# The hollow child, in its historical shape: a bare roster row
# with nothing inside — exactly what the round-8 void was made
# of. Shape-legal (so the seedable check passes) and arm-legal
# (no pair/caveat arm), so only the hollow-child check can
# refuse it.
{
"id": "X_t",
"arms": [ARM_NUDGE],
"children": [{"ws_id": "ws-c1", "name": "auditor", "state": "idle"}],
"tasks": [_OPEN_TASK],
},
# A world block with an unrecognized key — a silent no-op seed —
# trips only the world-shape check.
{
"id": "X_t",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"world": {"memroy": []},
},
]
def _tasks_call(action: str, turn: int = 0, **args: Any) -> dict[str, Any]:
"""One ``tasks`` tool-log entry, keyed with the tool's own primary
key so a rename of the action argument moves these fixtures too."""
return {
"tool": "tasks",
"args": {_TASKS_ACTION_KEY: action, **args},
"result": "",
"turn": turn,
}
def _cell(cell_id: str) -> dict[str, Any]:
return next(c for c in NUDGE_CELLS if c["id"] == cell_id)
def _envelope(*rows: dict[str, Any]) -> dict[str, Any]:
return {"version": 1, "tasks": list(rows)}
@pytest.fixture
def eval_storage(tmp_path):
reset_storage()
init_storage("sqlite", path=str(tmp_path / "eval.db"), run_migrations=False)
storage = get_storage()
storage.register_workstream(
"coord-eval-1",
user_id="eval-user",
name="eval-coordinator",
kind=WorkstreamKind.COORDINATOR,
state="idle",
)
yield storage
reset_storage()
class TestCellFixtures:
def test_nudge_cells_use_live_tool_names(self):
"""A forbid/expect matcher with a misspelt tool name silently
never matches — which reads as a false PASS. Every name must be
on the coordinator's real wire."""
for cell in NUDGE_CELLS:
named = [a["tool"] for a in cell.get("forbid_actions", [])]
named += [a["tool"] for a in cell.get("expect_actions", {}).get("actions", [])]
for name in named:
assert name in _LIVE_TOOL_NAMES, f"{cell['id']}: unknown tool {name!r}"
def test_cell_ids_unique_and_arms_known(self):
"""The arm vocabulary is imported, never re-declared: a cell's
own list is the only route into the grid, so a literal copy here
could drift from the runner's."""
ids = [c["id"] for c in NUDGE_CELLS]
assert len(ids) == len(set(ids))
for cell in NUDGE_CELLS:
assert set(cell["arms"]) <= KNOWN_ARMS, cell["id"]
def test_every_cell_seed_spec_passes_real_validation(self, eval_storage):
"""Fixture drift guard: every cell's tasks seed through the REAL
tasks_add (status vocabulary, length caps, renderability) — a
vocabulary change that orphans a fixture fails here, not mid-
sweep."""
for cell in NUDGE_CELLS:
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
id_map = _seed_tasks(client, "coord-eval-1", cell)
assert len(id_map) == len(cell.get("tasks", []))
assert all(tid.startswith("tsk_") for tid in id_map.values())
# Clean the envelope between cells (same coord ws).
env = client.tasks_get("coord-eval-1")
for row in env.get("tasks", []):
client.tasks_remove("coord-eval-1", task_id=row["id"])
def test_bad_seed_fixture_fails_loudly(self, eval_storage):
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
bad = {"id": "X", "tasks": [{"title": "t", "status": "not-a-status"}]}
with pytest.raises(ValueError, match="rejected"):
_seed_tasks(client, "coord-eval-1", bad)
class TestWorldSeeding:
"""``world`` seeds the TOOL-VISIBLE environment through production
writers, and the proof is the production READS: the memory rows come
back through the same listing the memory tool serves, and the node
comes back through the real ``CoordinatorClient.list_nodes`` — the
service-registry liveness intersection included."""
_WORLD_CELL = {
"id": "X_world",
"tasks": [{"title": "t", "status": "pending"}],
"world": {
"memory": [
{
"name": "proj-context",
"description": "Project deployment context",
"content": "acme-api: staging tracks main.",
"type": "reference",
}
],
"nodes": [{"node_id": "node-t", "metadata": {"hostname": "node-t", "os": "linux"}}],
},
}
def test_memory_rows_read_back_through_the_production_listing(self, eval_storage):
from turnstone.core.memory import list_structured_memories
_seed_world(eval_storage, self._WORLD_CELL)
rows = list_structured_memories(scope="global")
by_name = {r["name"]: r for r in rows}
# The production writer normalizes names (normalize_key), so the
# seeded row reads back exactly as a model-saved one would.
assert "proj_context" in by_name
assert by_name["proj_context"]["content"] == "acme-api: staging tracks main."
def test_nodes_read_back_through_the_real_list_nodes(self, eval_storage):
_seed_world(eval_storage, self._WORLD_CELL)
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
out = client.list_nodes()
ids = {n.get("node_id") for n in out.get("nodes", [])}
assert "node-t" in ids
# The FILTERED read is the sharp edge: filter values are
# JSON-encoded before comparison, so the seed must store them
# encoded exactly as production writers do or this matches
# nothing and the world is hollow for a filtering model.
filtered = client.list_nodes(filters={"os": "linux"})
assert {n.get("node_id") for n in filtered.get("nodes", [])} == {"node-t"}
def test_a_worldless_cell_seeds_nothing_and_raises_nothing(self, eval_storage):
from turnstone.core.memory import list_structured_memories
_seed_world(eval_storage, {"id": "X_plain", "tasks": []})
assert list_structured_memories(scope="global") == []
class TestSweepValidation:
"""Every fixture error a cell can carry is refused at config time —
before the canary probe spends a model round-trip — because each of
them otherwise surfaces as a plausible red 0% that is
indistinguishable in the result JSON from a real model failure.
Three shapes of harm: a stimulus that cannot be built (an unknown
arm), a run that dies mid-sweep on a dereference (a cell with no
id, a child with no ws_id, ``children: None``, a seed row with no
title, a transcript row with no role), and a run that completes but
is filed under a label it did not measure (a pair arm with no active
child, a caveat-ablation arm with no LIVE child, a body arm with no
open task, an ``expect_state`` index naming a task nobody seeded, a
repeated arm or cell id whose second batch overwrites the first).
"""
def test_shipped_cells_pass(self):
_validate_cells(list(NUDGE_CELLS))
def test_unknown_arm_is_refused_at_sweep_start(self):
"""The class E2 left open when it deleted the CLI override: with
the flag gone, a cell declaration is the ONLY way to name an arm
that does not exist."""
cell = {
"id": "X_typo",
"arms": [ARM_NUDGE, "pair_tff"],
"children": [],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_typo" in message, message
assert "pair_tff" in message, message
def test_unknown_arm_is_refused_before_the_pair_arm_check(self):
"""A misspelt PAIR arm is an unknown arm, not a childless pair
arm — the diagnostic must name the typo, not send the author
hunting for a child to seed."""
cell = {
"id": "X_typo_pair",
"arms": ["pair_ttf"],
"children": [],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "pair_ttf" in message and "unknown" in message, message
assert "active state" not in message, message
def test_arms_that_are_not_a_string_list_are_refused(self):
cell = {"id": "X_arms_scalar", "arms": ARM_NUDGE, "tasks": [_OPEN_TASK]}
with pytest.raises(SystemExit, match="X_arms_scalar"):
_validate_cells([cell])
def test_an_empty_arms_list_is_refused(self):
"""``all()`` over ``[]`` is vacuously true, so the shape refusal
passed a literal empty list and the cell swept as an empty
result — present in the file, contributing no runs to any rate,
read as a complete grid. Distinct from the ABSENT key, which
deliberately defaults to the plain nudge arm."""
cell = {"id": "X_no_arms", "arms": [], "tasks": [_OPEN_TASK]}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_no_arms" in message and "empty" in message, message
# The control for the "distinct from absent" clause.
_validate_cells([{"id": "X_default", "tasks": [_OPEN_TASK]}])
@pytest.mark.parametrize(
("row", "reject_fragment"),
[
# The [0] escape verbatim: the sibling pending row satisfies
# the open-task check, so only the write-path dry-run can
# refuse the cell.
({"title": "b", "status": "todo"}, "invalid status"),
({"title": "x" * 250}, "title too long"),
({"title": "b", "note": "n" * 250}, "note too long"),
# Renderability: zero-widths survive ``str.strip`` (they are
# not whitespace), so the title-presence guard passes and
# only ``tasks_add``'s sanitiser can reject the row.
({"title": chr(0x200B) * 3}, "renderable"),
],
)
def test_a_seed_row_the_write_path_rejects_is_refused(self, row, reject_fragment):
"""Validation BY CONSTRUCTION: the refusal set is ``tasks_add``'s
own, exercised through the real ``_seed_tasks`` against a
throwaway envelope. A hand-written restatement covered exactly
one of these four rows (none), and each miss was a plausible red
0% filed as a model result after the canary round-trip was paid.
"""
cell = {"id": "X_unseedable", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK, row]}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_unseedable" in message, message
assert reject_fragment in message, message
def test_tool_stubs_that_are_not_a_mapping_are_refused(self):
for stubs in (["not-a-mapping"], "wait_for_workstream", 7):
cell = {
"id": "X_stub_shape",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"tool_stubs": stubs,
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_stub_shape" in message and "mapping" in message, message
def test_a_tool_stub_queue_that_is_not_result_dicts_is_refused(self):
for queue in ({"complete": True}, ["a string result"], [{"ok": 1}, None]):
cell = {
"id": "X_stub_queue",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"tool_stubs": {"wait_for_workstream": queue},
}
with pytest.raises(SystemExit, match="X_stub_queue"):
_validate_cells([cell])
def test_well_shaped_tool_stubs_pass(self):
"""The refusal keys on the run path's consumed shape, so the
shape the run path consumes must pass."""
_validate_cells(
[
{
"id": "X_stub_ok",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"tool_stubs": {"wait_for_workstream": [{"complete": False, "results": {}}]},
}
]
)
def test_a_repeated_arm_is_refused(self):
"""The runner keys ``cell_out`` by arm name, so the second batch
overwrites the first: a full batch of live generations bought
and then discarded, with the file reporting half the runs it
paid for and no sign the rest existed."""
cell = {"id": "X_dupe_arm", "arms": [ARM_NUDGE, ARM_NUDGE], "tasks": [_OPEN_TASK]}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_dupe_arm" in message and ARM_NUDGE in message, message
def test_a_repeated_cell_id_is_refused(self):
"""Same overwrite mechanism one level up: ``out["cells"]`` is
keyed by cell id. ``test_cell_ids_unique_and_arms_known``
covers the SHIPPED list; this covers the sweep."""
cells = [
{"id": "X_twice", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]},
{"id": "X_twice", "arms": [ARM_BARE_CONTINUE], "tasks": [_OPEN_TASK]},
]
with pytest.raises(SystemExit, match="X_twice"):
_validate_cells(cells)
def test_a_cell_without_an_id_is_refused(self):
"""The runner dereferences ``case['id']`` per cell and per run,
so this kills the sweep mid-flight — tens of minutes of live
generation in, and before anything is written to --output."""
cells = [
{"id": "X_ok", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]},
{"arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]},
]
with pytest.raises(SystemExit) as excinfo:
_validate_cells(cells)
assert "position 1" in str(excinfo.value), str(excinfo.value)
def test_a_child_without_a_ws_id_is_refused(self):
cell = {
"id": "X_child_no_ws",
"arms": [ARM_PAIR_TF],
"children": [{"name": "auditor", "state": "running"}],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit, match="ws_id"):
_validate_cells([cell])
def test_children_declared_as_none_is_refused(self):
"""The validator's own ``or []`` used to coerce this away while
the run seeder iterated it unguarded — the validator passing a
cell that then dies on the first run of the sweep."""
cell = {
"id": "X_children_none",
"arms": [ARM_NUDGE],
"children": None,
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit, match="X_children_none"):
_validate_cells([cell])
def test_a_seed_task_without_a_title_is_refused(self):
cell = {"id": "X_task_no_title", "arms": [ARM_NUDGE], "tasks": [{"status": "pending"}]}
with pytest.raises(SystemExit, match="title"):
_validate_cells([cell])
def test_a_transcript_row_without_a_role_is_refused(self):
cell = {
"id": "X_no_role",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"transcript": [{"content": "hello"}],
}
with pytest.raises(SystemExit, match="role"):
_validate_cells([cell])
def test_a_seeded_tool_call_without_a_name_is_refused(self):
cell = {
"id": "X_no_call_name",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"transcript": [{"role": "assistant", "tool_calls": [{"args": {}}]}],
}
with pytest.raises(SystemExit, match="X_no_call_name"):
_validate_cells([cell])
def test_an_expect_state_index_with_no_seeded_task_is_refused(self):
"""The unseeded index maps to no task id, so every run scores
``task None missing from the final envelope`` — a full grid at
0% whose message reads as the coordinator having deleted a task
it was never given."""
cell = {
"id": "X_unseeded_index",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"expect_state": {1: {"status": "done"}},
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_unseeded_index" in message and "seed index 1" in message, message
def test_expect_state_indices_that_are_seeded_pass(self):
"""Both spellings the fixtures use — an int key and the string a
JSON round-trip leaves behind — must be accepted, or the refusal
would refuse real cells."""
for key in (0, "0"):
cell = {
"id": "X_seeded_index",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"expect_state": {key: {"status": "done"}},
}
_validate_cells([cell])
def test_a_cell_with_no_open_task_is_refused_for_body_arms(self):
"""``format_idle_tasks_nudge`` returns '' for an empty open list
and the runner appends the turn unconditionally, so the model
would get fence markers around an empty body — a wire production
never sends, with every run still filed under the nudge
heading."""
for status in ("done", "blocked", "needs_user"):
cell = {
"id": f"X_closed_{status}",
"arms": [ARM_NUDGE],
"tasks": [{"title": "already handled", "status": status}],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert f"X_closed_{status}" in message, message
assert "open status" in message, message
def test_the_body_arms_set_still_derives_by_subtraction(self):
"""The open-task refusal keys on ``_TASKS_BODY_ARMS``, which is
``KNOWN_ARMS`` minus the one no-body arm — so the ``nudge`` arm
(now the counts body's own measurement) stays covered through
the derivation, and an arm added to the vocabulary joins the
refusal unless someone explicitly excludes it. Pinned because
the arm retirements shrank the set by subtraction too, and a
hand-maintained list could have quietly dropped a survivor."""
from turnstone.eval.nudges import _TASKS_BODY_ARMS
assert KNOWN_ARMS - {ARM_BARE_CONTINUE} == _TASKS_BODY_ARMS
assert ARM_NUDGE in _TASKS_BODY_ARMS
def test_a_bare_continue_only_cell_needs_no_open_task(self):
"""The refusal is scoped to arms that RENDER a body: the
operator-poke baseline injects no nudge at all, so an empty open
list costs it nothing."""
_validate_cells(
[
{
"id": "X_poke_only",
"arms": [ARM_BARE_CONTINUE],
"tasks": [{"title": "already handled", "status": "done"}],
}
]
)
def test_a_forbid_matcher_without_a_tool_is_refused(self):
"""``_match_action`` reads ``expected['tool']`` with no default,
and it runs in the SCORER — so this shape dies after the run's
generations have already been bought."""
cell = {
"id": "X_matcher_no_tool",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"forbid_actions": [{"args_pattern": {"initial_message": "audit"}}],
}
with pytest.raises(SystemExit, match="X_matcher_no_tool"):
_validate_cells([cell])
def test_expect_actions_without_an_actions_list_is_refused(self):
cell = {
"id": "X_expect_shape",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"expect_actions": {"mode": "contains_any"},
}
with pytest.raises(SystemExit, match="X_expect_shape"):
_validate_cells([cell])
def test_known_arms_is_the_complete_arm_vocabulary(self):
"""The validator is only as good as this set. Both directions:
every ``ARM_*`` constant the module declares is in it (one added
without registering would be refused as an unknown arm), and
every member is an arm ``build_stimulus`` really accepts (one
dropped from the builder would pass validation and then raise
mid-sweep — the class the validator exists to kill)."""
from turnstone.eval import nudges as nudges_module
declared = {
value
for name, value in vars(nudges_module).items()
if name.startswith("ARM_") and isinstance(value, str)
}
assert declared == set(KNOWN_ARMS)
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
kids = [{"ws_id": "ws-c1", "name": "auditor", "state": "running"}]
for arm in sorted(KNOWN_ARMS):
assert build_stimulus(arm, envelope=env, children=kids), arm
def test_every_known_arm_is_accepted(self):
"""The refusal keys on :data:`KNOWN_ARMS`, so no legal arm may
trip it — an over-tight set would refuse real sweeps."""
for arm in sorted(KNOWN_ARMS):
cell = {
"id": f"X_{arm}",
"arms": [arm],
"children": [_LIVE_CHILD],
"tasks": [_OPEN_TASK],
}
_validate_cells([cell])
def test_pair_arm_on_childless_cell_is_refused_at_sweep_start(self):
# Iterates the module's own pair-arm set (one member since the
# ordering ablation retired) so a future pair variant joins this
# refusal coverage by registration, not by edit.
from turnstone.eval.nudges import _PAIR_ARMS
assert {ARM_PAIR_TF} == _PAIR_ARMS
for arm in sorted(_PAIR_ARMS):
cell = {
"id": f"X_childless_{arm}",
"arms": [arm],
"children": [],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert f"X_childless_{arm}" in message, message
assert "active state" in message, message
def test_pair_arm_with_only_inactive_children_is_refused(self):
"""``idle`` is precisely the C6b state: a child that exists but
that the observer would not nudge about. The child carries a
full transcript so the PAIR-ARM refusal is the only one this
cell can trip — a hollow child would trip its own check too and
the assertion would ride on registration order."""
cell = {
"id": "X_idle_child",
"arms": [ARM_NUDGE, ARM_PAIR_TF],
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": "idle",
"transcript": [_ASSIGNMENT_ROW, _FINDINGS_ROW],
}
],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
assert "X_idle_child" in str(excinfo.value)
def test_no_caveat_arm_on_a_childless_cell_is_refused(self):
"""The arm measures what the caveat buys WHERE IT PROTECTS. On a
childless cell it measures the childless body — which is the body
the conditional gives the plain ``nudge`` arm there anyway — so
the pair would read as an ablation result while being one
stimulus under two headings. Same honesty rule as the pair-arm
refusal."""
cell = {
"id": "X_no_caveat_childless",
"arms": [ARM_NUDGE, ARM_NO_CAVEAT],
"children": [],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_no_caveat_childless" in message, message
assert "live state" in message, message
def test_no_caveat_arm_with_only_terminal_children_is_refused(self):
"""``closed`` / ``deleted`` are the strings the close and reap
paths write, and they are NOT ``WorkstreamState`` members — a row
in one of them is gone, so the cell is childless for the body's
purposes however many rows it declares."""
for state in ("closed", "deleted"):
cell = {
"id": f"X_no_caveat_{state}",
"arms": [ARM_NO_CAVEAT],
# A full transcript, so the terminal-state refusal is
# the only one the cell can trip.
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": state,
"transcript": [_ASSIGNMENT_ROW, _FINDINGS_ROW],
}
],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
assert f"X_no_caveat_{state}" in str(excinfo.value), state
def test_no_caveat_arm_passes_on_any_live_child_including_idle(self):
"""EXISTENCE-in-a-live-state, deliberately broader than the pair
arms' ACTIVE predicate: C6b's idle child — a child that finished
with results nobody collected — is precisely the row the
stopped-child fact line protects, and it fails the pair-arm check
(``test_pair_arm_with_only_inactive_children_is_refused``) while
passing this one. Both are correct."""
for state in sorted(_LIVE_CHILD_STATES):
cell = {
"id": f"X_no_caveat_live_{state}",
"arms": [ARM_NUDGE, ARM_NO_CAVEAT],
# Each state's LEGAL transcript: assignment always, and
# the completion turn only where the hollow-child check
# demands one (idle). error deliberately gets the
# assignment alone — errored-before-output is a
# production-reachable world, pinned separately by
# ``test_an_error_child_is_not_bound_by_the_findings_rule``.
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": state,
"transcript": [_ASSIGNMENT_ROW]
+ ([_FINDINGS_ROW] if state == "idle" else []),
}
],
"tasks": [_OPEN_TASK],
}
_validate_cells([cell])
def test_a_childless_cell_is_refused_under_an_override(self):
"""Item: an override sweep can no longer force children content
into a childless world, STRUCTURALLY. A childless cell renders
the formatter's childless branch, whose literal door cut would
silently strip a candidate that quotes the shipped
blocked-on-a-child branch — so the sweep refuses the cell at
config validation, naming the cell and the reason, before any
model round-trip. Without an override the same cell passes:
the shipped tail is exactly what the cut is defined against.
"""
cell = {
"id": "X_override_childless",
"arms": [ARM_NUDGE],
"children": [],
"tasks": [_OPEN_TASK],
}
_validate_cells([cell]) # fine against the shipped body
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell], override_active=True)
message = str(excinfo.value)
assert "X_override_childless" in message, message
assert "body-override" in message, message
assert "live state" in message, message
def test_a_terminal_only_children_cell_is_refused_under_an_override(self):
"""The refusal keys on the LIVE derivation, not the raw list: a
cell whose every child row is terminal is a childless world to
the formatter, so a raw-list predicate would wave it through and
the door cut would maul the candidate anyway."""
cell = {
"id": "X_override_terminal",
"arms": [ARM_NUDGE],
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": "closed",
"transcript": [_ASSIGNMENT_ROW, _FINDINGS_ROW],
}
],
"tasks": [_OPEN_TASK],
}
_validate_cells([cell])
with pytest.raises(SystemExit, match="X_override_terminal"):
_validate_cells([cell], override_active=True)
def test_a_live_child_cell_passes_under_an_override(self):
"""The refusal is exactly as wide as the maul: any live-state
child row keeps the door in play, so the cell measures the
candidate as authored (idle included — the C6b class is a legal
override cell)."""
for state in ("running", "idle"):
cell = {
"id": f"X_override_{state}",
"arms": [ARM_NUDGE],
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": state,
"transcript": [_ASSIGNMENT_ROW]
+ ([_FINDINGS_ROW] if state == "idle" else []),
}
],
"tasks": [_OPEN_TASK],
}
_validate_cells([cell], override_active=True)
def test_the_shipped_cells_that_survive_an_override_are_the_children_ones(self):
"""The shipped grid under ``--body-override``: exactly the two
children-bearing cells validate; every childless cell is refused
by name. (An override sweep therefore runs with ``--cells
C6_co_delivery,C6b_stranded_children`` or a fixture edit — never
with a silently mauled childless body.)"""
with_children = [c for c in NUDGE_CELLS if _live_children(c.get("children") or [])]
assert {c["id"] for c in with_children} == {"C6_co_delivery", "C6b_stranded_children"}
_validate_cells(with_children, override_active=True)
for cell in NUDGE_CELLS:
if cell in with_children:
continue
with pytest.raises(SystemExit, match=cell["id"]):
_validate_cells([cell], override_active=True)
def test_the_cells_that_declare_no_caveat_are_the_ones_with_children(self):
"""The shipped grid, named rather than assumed.
The subset direction is the honesty rule the validator enforces
for any cell (an arm that cannot measure anything is refused);
the two ids are this sweep's choice, and they are the two cells
that seed a live child today — C6's running one and C6b's idle
one, the pair that covers both disjuncts of the sentence. A
third children-bearing cell may decline the arm (it costs a full
batch of live generations), which is why only the subset is
asserted in that direction.
"""
declaring = {c["id"] for c in NUDGE_CELLS if ARM_NO_CAVEAT in c.get("arms", [])}
with_children = {c["id"] for c in NUDGE_CELLS if _live_children(c.get("children") or [])}
assert declaring == {"C6_co_delivery", "C6b_stranded_children"}
assert declaring <= with_children
def test_missing_children_key_is_refused(self):
cell = {"id": "X_no_children_key", "arms": [ARM_PAIR_TF], "tasks": [_OPEN_TASK]}
with pytest.raises(SystemExit, match="X_no_children_key"):
_validate_cells([cell])
def test_cell_with_an_active_child_passes(self):
for state in sorted(_ACTIVE_CHILD_STATES):
cell = {
"id": "X_active",
"arms": [ARM_PAIR_TF, ARM_NUDGE],
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": state,
"transcript": [_ASSIGNMENT_ROW],
}
],
"tasks": [_OPEN_TASK],
}
_validate_cells([cell])
def test_a_stateless_child_is_seeded_active_and_accepted(self):
"""The disagreement that was a live trap: the run seeder
registers a child with no ``state`` as running (an ACTIVE
state), while the validator's filter read the missing key as
``None`` and refused the cell — a diagnostic contradicting the
state the run would have seeded. One named default, read
through one accessor, is what makes them agree; the stimulus
must render the same child too."""
cell = {
"id": "X_stateless_child",
"arms": [ARM_PAIR_TF],
"children": [{"ws_id": "ws-c1", "name": "auditor", "transcript": [_ASSIGNMENT_ROW]}],
"tasks": [_OPEN_TASK],
}
_validate_cells([cell])
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
turns = build_stimulus(ARM_PAIR_TF, envelope=env, children=cell["children"])
assert [t.get("_source") for t in turns[1:]] == ["idle_tasks", "idle_children"]
assert "running" in turns[2]["content"], turns[2]["content"]
def test_a_child_transcript_that_is_not_a_list_is_refused(self):
"""The transcript seeder iterates it unguarded — the same
dereference-mirror rule as ``children: None``."""
cell = {
"id": "X_transcript_scalar",
"arms": [ARM_NUDGE],
"children": [{"ws_id": "ws-c1", "state": "running", "transcript": "do the audit"}],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_transcript_scalar" in message and "not a list" in message, message
def test_a_child_transcript_row_without_role_or_content_is_refused(self):
"""The seeder dereferences both keys directly; a row missing
either dies in the harness on every run of the cell."""
for row in ({"content": "no role"}, {"role": "assistant"}, "not-a-row"):
cell = {
"id": "X_transcript_row",
"arms": [ARM_NUDGE],
"children": [{"ws_id": "ws-c1", "state": "running", "transcript": [row]}],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_transcript_row" in message and "role" in message, message
def test_a_hollow_child_is_refused(self):
"""The round-8 shape, refused at sweep start: a bare roster row
with nothing inside. ``inspect`` is real in every run, so any
cell's model can look inside a child and find ``messages: []``
contradicting the roster — a world ``spawn`` cannot produce
(it writes the assignment before the child ever runs). Both
the absent key and the explicit empty list are the same
hollowness."""
for transcript_shape in ({}, {"transcript": []}):
cell = {
"id": "X_hollow_child",
"arms": [ARM_NUDGE],
"children": [
{"ws_id": "ws-c1", "name": "auditor", "state": "running", **transcript_shape}
],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_hollow_child" in message, message
assert "seeds no transcript" in message, message
assert "assignment" in message, message
def test_an_idle_child_without_findings_is_refused(self):
"""The C6b void, refused at sweep start: idle resolves the
synthesized wait ``complete`` and the wait carries the child's
last assistant message — an idle child without one tells the
model the work finished and shows nothing was produced, so it
correctly redoes the work and the forbidden rate measures the
fixture. The refusal is asked of the REAL reader, so an
assistant row whose content is whitespace is as hollow as no
assistant row at all."""
for transcript in (
[_ASSIGNMENT_ROW],
[_ASSIGNMENT_ROW, {"role": "assistant", "content": " "}],
):
cell = {
"id": "X_idle_no_findings",
"arms": [ARM_NUDGE],
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": "idle",
"transcript": transcript,
}
],
"tasks": [_OPEN_TASK],
}
with pytest.raises(SystemExit) as excinfo:
_validate_cells([cell])
message = str(excinfo.value)
assert "X_idle_no_findings" in message, message
assert "surfaces" in message and "assistant" in message, message
def test_a_running_child_with_only_its_assignment_passes(self):
"""The mid-work world: assignment, no output yet. The wait
synthesis answers ``message: None`` for a running child by
production's own rule, so nothing about this shape lies."""
_validate_cells(
[
{
"id": "X_mid_work",
"arms": [ARM_NUDGE],
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": "running",
"transcript": [_ASSIGNMENT_ROW],
}
],
"tasks": [_OPEN_TASK],
}
]
)
def test_an_error_child_is_not_bound_by_the_findings_rule(self):
"""``error`` is terminal like ``idle``, but errored-before-output
is a production-REACHABLE world (``_last_assistant_text``
documents it) and the wait surfaces the persisted ``last_error``
or the honest sentinel — so an error child with only its
assignment is a legal fixture, while a message-less one is still
refused by the transcript rule."""
_validate_cells(
[
{
"id": "X_error_child",
"arms": [ARM_NUDGE],
"children": [
{
"ws_id": "ws-c1",
"name": "auditor",
"state": "error",
"transcript": [_ASSIGNMENT_ROW],
}
],
"tasks": [_OPEN_TASK],
}
]
)
def test_non_pair_cells_need_no_children(self):
_validate_cells(
[
{"id": "X_plain", "arms": [ARM_NUDGE, ARM_BARE_CONTINUE], "tasks": [_OPEN_TASK]},
{"id": "X_default_arm", "tasks": [_OPEN_TASK]},
]
)
def test_world_memory_row_missing_content_is_refused(self):
"""The field refusals are the only guard in front of
``_seed_world``'s bare indexing — a regression here becomes a
mid-sweep KeyError after the canary spends round-trips. Own
test, NOT a ``_trip_cells`` entry: that helper is zipped
``strict=True`` against the check table."""
import re
cells = [
{
"id": "X_world_bad",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"world": {"memory": [{"name": "x"}]},
}
]
with pytest.raises(SystemExit, match=re.escape("world.memory[0].content")):
_validate_cells(cells)
def test_world_node_row_with_empty_id_is_refused(self):
import re
cells = [
{
"id": "X_world_bad",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK],
"world": {"nodes": [{"node_id": ""}]},
}
]
with pytest.raises(SystemExit, match=re.escape("world.nodes[0].node_id")):
_validate_cells(cells)
def test_a_later_bad_cell_still_refuses(self):
"""The scan covers the whole list, not just its head."""
for bad in (
{"arms": [ARM_PAIR_TF], "children": [], "tasks": [_OPEN_TASK]},
{"arms": ["not_an_arm"], "tasks": [_OPEN_TASK]},
):
cells = [
{"id": "X_ok", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]},
{"id": "X_bad", **bad},
]
with pytest.raises(SystemExit, match="X_bad"):
_validate_cells(cells)
def test_no_refusal_is_reachable_only_behind_another_ones_early_out(self):
"""The structural property the check table buys.
The pair-arm class used to sit behind ``if not declared:
continue``, so anything appended after it did not run for cells
declaring no pair arm — exactly the cell class a new refusal is
most likely to be about. Asserted on the artefact: every
registered check fires for a cell that trips ONLY it, whatever
its position in the table.
"""
from turnstone.eval import nudges as nudges_module
trips = _trip_cells()
assert len(trips) == len(nudges_module._CELL_CHECKS), (
"every registered check needs a cell that trips only it"
)
for i, (check, cell) in enumerate(zip(nudges_module._CELL_CHECKS, trips, strict=True)):
shadowed = [c.__name__ for c in nudges_module._CELL_CHECKS[:i] if c(cell) is not None]
assert not shadowed, f"{check.__name__}'s cell is refused earlier by {shadowed}"
assert check(cell) is not None, check.__name__
with pytest.raises(SystemExit, match="X_t"):
_validate_cells([cell])
class TestRefusalsPrecedeTheCanary:
"""A bad fixture must cost ZERO model round-trips.
``_validate_cells`` runs before ``tool_call_canary``, and the probe
is the first thing in the sweep that touches the endpoint. Proved
on the artefact rather than by reading the call order: the probe is
replaced with a sentinel that raises on ENTRY, so a refusal that
reached it would surface as the sentinel instead of the ``SystemExit``.
Both directions, because "everything raises before the probe" is
also what a validator that refuses every cell would produce: each
bad cell must raise ``SystemExit`` WITHOUT the sentinel, and a good
cell must raise the sentinel — i.e. really get that far.
"""
_SENTINEL = "canary-probe-entered"
@classmethod
def _sweep(
cls, monkeypatch, cells: list[dict[str, Any]], *, override: str | None = None
) -> None:
from turnstone.eval import nudges as nudges_module
def _probe_sentinel(*a: Any, **k: Any) -> bool:
raise RuntimeError(cls._SENTINEL)
def _never_runs(**kw: Any) -> dict[str, Any]:
raise AssertionError("a run reached the model lane")
monkeypatch.setattr(nudges_module, "tool_call_canary", _probe_sentinel)
monkeypatch.setattr(nudges_module, "_run_single_nudge", _never_runs)
run_nudge_response(
base_url="http://eval.invalid/v1",
api_key="x",
model="eval-model",
cells=cells,
n_runs=1,
body_override_text=override,
)
def test_a_good_cell_reaches_the_probe(self, monkeypatch):
"""The control. Without it, every assertion below would also
pass against a validator that refused everything."""
cell = {"id": "X_good", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]}
with pytest.raises(RuntimeError, match=self._SENTINEL):
self._sweep(monkeypatch, [cell])
def test_the_shipped_cells_reach_the_probe(self, monkeypatch):
with pytest.raises(RuntimeError, match=self._SENTINEL):
self._sweep(monkeypatch, list(NUDGE_CELLS))
def test_every_refusal_fires_before_the_probe(self, monkeypatch):
bad: list[dict[str, Any]] = [
*_trip_cells(),
{"arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]}, # no id
# The construction-based seed refusals: an unseedable status,
# an over-cap title, an unrenderable title. Each must cost
# zero model round-trips, like every other refusal.
{
"id": "X_t",
"arms": [ARM_NUDGE],
"tasks": [_OPEN_TASK, {"title": "b", "status": "todo"}],
},
{"id": "X_t", "arms": [ARM_NUDGE], "tasks": [{"title": "x" * 250}]},
{"id": "X_t", "arms": [ARM_NUDGE], "tasks": [{"title": chr(0x200B) * 3}]},
]
for cell in bad:
with pytest.raises(SystemExit) as excinfo:
self._sweep(monkeypatch, [cell])
assert self._SENTINEL not in str(excinfo.value), cell
def test_the_override_refusal_fires_before_the_probe(self, monkeypatch):
"""The override-only class binds at the same point as every
other refusal: config time, zero model round-trips. Both
directions, like the class docstring demands — the childless
cell refuses WITHOUT the sentinel, and the same cell without an
override reaches the probe (so the refusal really is
override-conditional)."""
cell = {"id": "X_t", "arms": [ARM_NUDGE], "children": [], "tasks": [_OPEN_TASK]}
with pytest.raises(SystemExit) as excinfo:
self._sweep(monkeypatch, [cell], override="Candidate wording.")
assert self._SENTINEL not in str(excinfo.value)
with pytest.raises(RuntimeError, match=self._SENTINEL):
self._sweep(monkeypatch, [cell])
def test_a_duplicate_cell_id_fires_before_the_probe(self, monkeypatch):
"""The one refusal that reads the whole list rather than a
single cell, so it needs its own two-cell fixture."""
dupes = [
{"id": "X_twice", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]},
{"id": "X_twice", "arms": [ARM_BARE_CONTINUE], "tasks": [_OPEN_TASK]},
]
with pytest.raises(SystemExit) as excinfo:
self._sweep(monkeypatch, dupes)
assert self._SENTINEL not in str(excinfo.value)
class TestStimulus:
def test_arm_turn_shapes(self):
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
kids = [{"ws_id": "ws-c1", "name": "auditor", "state": "running"}]
nudge = build_stimulus(ARM_NUDGE, envelope=env, children=kids)
assert [t["role"] for t in nudge] == ["user", "system"]
assert nudge[0]["content"] == "" and nudge[0]["_source"] == "system_nudge"
assert nudge[1]["_source"] == "idle_tasks"
# Same single-turn shape as ``nudge`` — the two arms differ by
# one ARGUMENT to the production formatter, nothing else.
no_caveat = build_stimulus(ARM_NO_CAVEAT, envelope=env, children=kids)
assert [t["role"] for t in no_caveat] == ["user", "system"]
assert no_caveat[1]["_source"] == "idle_tasks"
tf = build_stimulus(ARM_PAIR_TF, envelope=env, children=kids)
assert [t.get("_source") for t in tf[1:]] == ["idle_tasks", "idle_children"]
bare = build_stimulus(ARM_BARE_CONTINUE, envelope=env, children=[])
assert bare == [{"role": "user", "content": "continue"}]
with pytest.raises(ValueError):
build_stimulus("nope", envelope=env, children=[])
def test_pair_arms_omit_the_children_turn_when_no_child_is_active(self):
"""An empty children body is a turn production never sends: the
observer short-circuits on empty text before enqueueing, and an
appended turn would fence open/close markers around nothing.
"No active child" covers both no children at all and children
whose state is outside the observer's active set."""
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
none_at_all = build_stimulus(ARM_PAIR_TF, envelope=env, children=[])
assert [t.get("_source") for t in none_at_all[1:]] == ["idle_tasks"]
idle_only = build_stimulus(
ARM_PAIR_TF,
envelope=env,
children=[{"ws_id": "ws-c1", "name": "auditor", "state": "idle"}],
)
assert [t.get("_source") for t in idle_only[1:]] == ["idle_tasks"]
def test_children_turn_filters_on_the_observers_own_state_set(self):
"""The active-state filter is the observer's frozenset, not a
copy of its membership — every state IN it renders the turn."""
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
for state in sorted(_ACTIVE_CHILD_STATES):
kids = [{"ws_id": "ws-c1", "name": "auditor", "state": state}]
turns = build_stimulus(ARM_PAIR_TF, envelope=env, children=kids)
assert [t.get("_source") for t in turns[1:]] == ["idle_tasks", "idle_children"], state
def test_bodies_are_production_rendered(self):
"""The eval carries no copy of the body: the real formatter over
the really-seeded envelope — the counts opener, the per-child
observed fact line, the id block and the escape branch,
populated with the seeded id and the seeded child, and NO task
text."""
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
body = render_tasks_body(env, children=[("ws-c1", "running")])
assert body.startswith("You still have 1 open task: 0 in_progress, 1 pending.")
assert NUDGE_CHILD_RUNNING_LINE.format(ws_id="ws-c1").removeprefix(chr(10)) in body
assert "needs_user" in body
# The seeded id reaches the block AND the branch calls; the
# seeded title reaches neither.
assert chr(10) + " - tsk_1 (pending)" in body
assert "task_id='tsk_1'" in body
assert "child_ws_id='ws-c1'" in body
assert "audit auth.py" not in body
def test_the_no_caveat_arm_is_the_formatters_other_branch(self):
"""No string surgery: the arm's body is what the production
formatter renders for a caller that observed no children, so the
ablation cannot drift from the body that ships.
THE ARM ABLATES THE BODY'S WHOLE CHILDREN AWARENESS, by design.
The ``children`` pairs govern the per-child fact lines, the
blocked-on-a-child branch and that branch's slots, because in
production all of them answer ONE storage read and a second
derivation of it is a state where two answers can disagree. So
the arm asks a single clean question — does this body need to
mention children at all? — rather than the narrower "what does
the sentence buy?" it was named for. That is a better question,
which is why the arm is worth keeping; the name stays because
archived sweeps report under it.
This scope is not drift. It is what "the formatter's other
branch" MEANS now that one read governs every children-bearing
element, and the alternative — a knob that stripped the fact
lines while keeping the branch — would render a body no
coordinator receives, which is the one thing an eval arm may
never do.
"""
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
kept = render_tasks_body(env, children=[("ws-c1", "running")])
ablated = render_tasks_body(env, children=[])
assert "Child ws-c1 is still running" in kept
assert "child_ws_id='ws-c1'" in kept
# NOTHING about children survives the ablation — asserted as
# absence of the topic, so a reworded branch cannot pass.
for absent in ("child", "Child", "wait_for_workstream", "list_workstreams"):
assert absent not in ablated, absent
# ...and nothing else moves. The counts opener, the id block and
# the remaining branches survive: this arm ablates an
# observation, not the body.
assert ablated.startswith("You still have 1 open task")
assert "needs_user" in ablated
assert "task_id='tsk_1'" in ablated
assert chr(10) + " - tsk_1 (pending)" in ablated
turns = build_stimulus(ARM_NO_CAVEAT, envelope=env, children=[_LIVE_CHILD])
assert turns[1]["content"] == ablated
def test_the_nudge_arm_derives_the_children_facts_from_the_cells_children(self):
"""The ``nudge`` arm renders what a coordinator in that cell's
state would really receive, which since the observer's probe
landed is two different bodies by cell class — and for the
children-bearing class, the OBSERVED per-state fact lines.
The predicate is EXISTENCE in a live state, the observer's own:
an idle child keeps its stopped-with-immediate-wait line (it
may hold results nobody collected) while a terminal row does not
count as a child at all. Deriving through ``_live_children``
rather than the pair arms' active filter is what those two rows
detect — and the STATE riding beside the id is what makes the
idle row render as the fact it is instead of a running claim or
a hedge.
"""
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
def _body(children: list[dict[str, str]]) -> str:
turns = build_stimulus(ARM_NUDGE, envelope=env, children=children)
return str(turns[1]["content"])
assert "Child " not in _body([])
assert NUDGE_CHILD_RUNNING_LINE.format(ws_id="ws-c1").removeprefix(chr(10)) in _body(
[_LIVE_CHILD]
)
idle_body = _body([{"ws_id": "ws-c1", "name": "auditor", "state": "idle"}])
assert (NUDGE_CHILD_STOPPED_LINE.format(ws_id="ws-c1").removeprefix(chr(10))) in idle_body
assert "Child " not in _body([{"ws_id": "ws-c1", "name": "auditor", "state": "closed"}])
# No hedge about an observed state, in any cell class.
for children in ([], [_LIVE_CHILD], [{"ws_id": "ws-c1", "state": "idle"}]):
body = _body(children)
assert "may still be running" not in body, children
assert "while you worked" not in body, children
def test_ragged_envelope_rows_do_not_raise(self):
env = _envelope(
{"id": "tsk_1", "title": None, "status": "pending", "note": 42},
"not-a-dict-row",
)
body = render_tasks_body(env, children=[("ws-c1", "running")])
assert body.startswith("You still have 1 open task")
def test_seed_transcript_pairs_calls_with_results(self):
wires = _seed_transcript(_cell("C6_co_delivery"))
roles = [w["role"] for w in wires]
assert roles == ["user", "assistant", "tool"]
call = wires[1]["tool_calls"][0]
assert call["function"]["name"] == "spawn_workstream"
assert json.loads(call["function"]["arguments"])["name"] == "auditor"
assert wires[2]["tool_call_id"] == call["id"]
_SKIPPED_ARMS = [
(ARM_NO_CAVEAT, _NO_CAVEAT_SKIP_REASON),
]
_SKIPPED_ARM_NAMES = [arm for arm, _reason in _SKIPPED_ARMS]
class TestBodyOverrideSkip:
"""The ablation arm is defined against the body that SHIPS, and
``--body-override`` replaces that body with unknown text.
``no_caveat`` cuts a LITERAL, so the failure against candidate text
is silent in the worst way: the likeliest candidate of all, one
that keeps the caveat sentence verbatim while rewording another
paragraph, still contains it, so the cut lands and the sweep files
a caveat-stripped candidate under the tuning heading — a real
number reported against a stimulus nobody chose. (The retired
``no_provenance`` arm had the positional variant of this failure;
it died with the paragraph it measured.)
The runner skips the arm instead of refusing the sweep, because a
cell's arm list is the only route into the grid — exiting would kill
every tuning sweep over the cells that declare it.
"""
@staticmethod
def _grid(
monkeypatch, *, override: str | None, arm: str = ARM_NO_CAVEAT
) -> tuple[dict[str, Any], list[str]]:
from turnstone.eval import nudges as nudges_module
ran: list[str] = []
def _fake_run(**kw: Any) -> dict[str, Any]:
ran.append(kw["arm"])
return {"pass": True, "failures": [], "forbidden": [], "actions": ["tasks"]}
monkeypatch.setattr(nudges_module, "tool_call_canary", lambda *a, **k: True)
monkeypatch.setattr(nudges_module, "_run_single_nudge", _fake_run)
out = run_nudge_response(
base_url="http://eval.invalid/v1",
api_key="x",
model="eval-model",
cells=[
{
"id": "X_tuning",
"arms": [ARM_NUDGE, arm],
# The live child is load-bearing for ``no_caveat``:
# without it the sweep-start validator refuses the
# cell and the SystemExit lands before any skip logic
# runs, which would read as a validator bug rather
# than as this fixture missing a row.
"children": [_LIVE_CHILD],
"tasks": [_OPEN_TASK],
}
],
n_runs=2,
body_override_text=override,
)
return out["cells"]["X_tuning"], ran
@pytest.mark.parametrize(("arm", "reason"), _SKIPPED_ARMS)
def test_an_ablation_arm_with_body_override_is_skipped(self, monkeypatch, capsys, arm, reason):
cell, ran = self._grid(
monkeypatch, override="Candidate wording, no paragraph break.", arm=arm
)
assert ran == [ARM_NUDGE, ARM_NUDGE], "the ablation arm must not reach a model"
assert cell[arm]["runs"] == []
assert reason in capsys.readouterr().out
@pytest.mark.parametrize(("arm", "reason"), _SKIPPED_ARMS)
def test_the_skip_keeps_the_uniform_per_arm_shape(self, monkeypatch, arm, reason):
"""Every archived result file carries the same four per-arm keys
and consumers iterate them, so the skip may ADD a key and never
drop one. Zero runs with null rates is what marks it as a
non-measurement — a real arm never reports either."""
cell, _ran = self._grid(
monkeypatch, override="Candidate wording, no paragraph break.", arm=arm
)
measured, skipped = cell[ARM_NUDGE], cell[arm]
assert set(skipped) == set(measured) | {"skipped"}
assert skipped == {
"n": 0,
"pass_rate": None,
"forbidden_rate": None,
"runs": [],
"skipped": reason,
}
assert json.loads(json.dumps(cell)) == cell # the grid is still writable
@pytest.mark.parametrize("arm", _SKIPPED_ARM_NAMES)
def test_an_ablation_arm_runs_when_no_override_is_in_play(self, monkeypatch, arm):
"""The guard must not cost the production sweep its arm: with no
override the ablation is against the shipped body, which is
exactly what it measures."""
cell, ran = self._grid(monkeypatch, override=None, arm=arm)
assert ran == [ARM_NUDGE, ARM_NUDGE, arm, arm]
assert "skipped" not in cell[arm]
assert cell[arm]["n"] == 2
def test_a_children_cell_measures_a_door_quoting_candidate_as_authored(self):
"""The LIKELIEST candidate shape there is: one that keeps the
shipped blocked-on-a-child branch verbatim while rewording
another paragraph. On a children cell — the only cell class an
override sweep still admits — the formatter's door cut never
runs (children are present), so the candidate's quoted branch
survives byte-exact and its slots are populated exactly as
production would populate the shipped text. The childless cell
that would have mauled this candidate is refused at config time
(``test_a_childless_cell_is_refused_under_an_override``), which
is the structural half of the same protection.
Asserted through the REAL override context manager, because the
mechanism under test is the one that swaps the module constant.
"""
from turnstone.eval import nudges as nudges_module
env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"})
candidate = "Reworded opening paragraph: reconcile your list." + NUDGE_IDLE_TASKS_CHILD_DOOR
with nudges_module._body_override(candidate):
rendered = render_tasks_body(env, children=[("ws-c1", "running")])
through_the_arm = build_stimulus(ARM_NUDGE, envelope=env, children=[_LIVE_CHILD])
# The control: the childless branch really would cut the
# quoted door out of this candidate — the maul the refusal
# and the skip exist to make unreachable.
mauled = render_tasks_body(env, children=[])
# Formatter-built facts lead; the candidate rides after them
# with its quoted door substituted, never stripped.
assert rendered.startswith(
"You still have 1 open task: 0 in_progress, 1 pending."
+ chr(10)
+ NUDGE_CHILD_RUNNING_LINE.format(ws_id="ws-c1").removeprefix(chr(10))
)
assert "record the link and wait instead of redoing its work" in rendered
assert "child_ws_id='ws-c1'" in rendered
assert NUDGE_IDLE_TASKS_CHILD_SLOT not in rendered
# The candidate's own task-id slot has no open-list machinery in
# this candidate, so it substitutes from the seeded set like the
# shipped tail's would.
assert "task_id='tsk_1'" in rendered
assert NUDGE_IDLE_TASKS_ID_SLOT not in rendered
assert through_the_arm[1]["content"] == rendered
assert "record the link and wait instead of redoing its work" not in mauled, (
"the control must show the cut really lands on candidate text"
)
class TestBodyFingerprint:
"""A result file must say WHICH body produced its numbers.
The ``nudge`` heading named one stimulus per cell only while the
caveat was unconditional; now that it is conditioned on an observed
fact the heading covers two, by cell class. None of the archived
result files records body text, hash or revision, so without this
the break between an archived sweep and a later one is folklore.
"""
@staticmethod
def _sweep(
monkeypatch, *, override: str | None, cells: list[dict[str, Any]] | None = None
) -> dict[str, Any]:
from turnstone.eval import nudges as nudges_module
monkeypatch.setattr(nudges_module, "tool_call_canary", lambda *a, **k: True)
monkeypatch.setattr(
nudges_module,
"_run_single_nudge",
lambda **kw: {"pass": True, "failures": [], "forbidden": [], "actions": ["tasks"]},
)
return run_nudge_response(
base_url="http://eval.invalid/v1",
api_key="x",
model="eval-model",
cells=cells
if cells is not None
else [
{"id": "X_a", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]},
{
"id": "X_b",
"arms": [ARM_NUDGE],
"children": [_LIVE_CHILD],
"tasks": [_OPEN_TASK],
},
],
n_runs=1,
body_override_text=override,
)
def test_the_fingerprint_names_the_shipped_tail_and_every_cell(self, monkeypatch):
from turnstone.core import metacognition as metacog
out = self._sweep(monkeypatch, override=None)
assert out["body"]["override"] is False
assert (
out["body"]["tail_sha256"]
== hashlib.sha256(metacog.NUDGE_IDLE_TASKS_TAIL.encode()).hexdigest()
)
assert set(out["body"]["cells"]) == {"X_a", "X_b"}
# The two cell classes the one ``nudge`` heading now covers,
# stamped by class: ``X_a`` seeds no children and ``X_b`` seeds a
# live one, so the fact is read off the cells rather than
# asserted flat, and a stamp that stopped tracking the stimulus
# fails here.
assert out["body"]["cells"] == {
"X_a": {"children_present": False},
"X_b": {"children_present": True},
}
assert json.loads(json.dumps(out["body"])) == out["body"]
def test_the_fingerprint_is_of_the_effective_tail_under_an_override(self, monkeypatch):
"""Taken INSIDE the override context, so the hash is of the text
the runs really saw — a fingerprint of the shipped tail under a
candidate sweep would be a false provenance stamp, which is
worse than none.
An override sweep admits only children-bearing cells (the
childless class is refused at config time), so the default
two-cell fixture cannot serve here — both cells carry a live
child, and both stamp the ``children_present`` fact their runs
really rendered, DERIVED rather than forced by a flag.
"""
from turnstone.core import metacognition as metacog
candidate = "Candidate wording under test."
out = self._sweep(
monkeypatch,
override=candidate,
cells=[
{
"id": "X_b",
"arms": [ARM_NUDGE],
"children": [_LIVE_CHILD],
"tasks": [_OPEN_TASK],
},
{
"id": "X_c",
"arms": [ARM_NUDGE],
"children": [
{
"ws_id": "ws-c9",
"name": "researcher",
"state": "idle",
"transcript": [
_ASSIGNMENT_ROW,
_FINDINGS_ROW,
],
}
],
"tasks": [_OPEN_TASK],
},
],
)
assert out["body"]["override"] is True
assert out["body"]["tail_sha256"] == hashlib.sha256(candidate.encode()).hexdigest()
assert out["body"]["cells"] == {
"X_b": {"children_present": True},
"X_c": {"children_present": True},
}
# ...and the constant is restored afterwards, so the stamp is
# not evidence of a leaked override.
assert candidate != metacog.NUDGE_IDLE_TASKS_TAIL
def test_the_fingerprint_stays_out_of_the_arm_mapping(self, monkeypatch):
"""Consumers iterate ``cells[id]`` as arms, so a non-arm key
there would read as an arm that reported no runs."""
out = self._sweep(monkeypatch, override=None)
for cell in out["cells"].values():
assert set(cell) <= KNOWN_ARMS, cell
class TestScoring:
def test_forbidden_action_flags_without_failing_state(self):
case = {"forbid_actions": [{"tool": "spawn_workstream"}]}
log = [{"tool": "spawn_workstream", "args": {"name": "x"}, "result": "", "turn": 0}]
r = score_nudge_run(log, _envelope(), case, {})
assert not r["pass"] and r["forbidden"] and not r["failures"]
def test_expect_state_checks_status_note_and_link(self):
case = {
"expect_state": {
0: {"status": "needs_user", "note_nonempty": True},
1: {"status": "in_progress", "child_ws_id": "ws-c1"},
}
}
good = _envelope(
{"id": "tsk_a", "status": "needs_user", "note": "need the token"},
{"id": "tsk_b", "status": "in_progress", "child_ws_id": "ws-c1"},
)
r = score_nudge_run([], good, case, {0: "tsk_a", 1: "tsk_b"})
assert r["pass"], r["failures"]
bad = _envelope(
{"id": "tsk_a", "status": "needs_user", "note": " "},
{"id": "tsk_b", "status": "in_progress", "child_ws_id": ""},
)
r2 = score_nudge_run([], bad, case, {0: "tsk_a", 1: "tsk_b"})
assert not r2["pass"] and len(r2["failures"]) == 2
def test_missing_task_is_a_state_failure(self):
case = {"expect_state": {0: {"status": "done"}}}
r = score_nudge_run([], _envelope(), case, {0: "tsk_gone"})
assert not r["pass"] and "missing" in r["failures"][0]
def test_require_stop_flags_work_after_bookkeeping(self):
case = {"require_stop": True}
log = [
_tasks_call("update"),
{"tool": "spawn_workstream", "args": {}, "result": "", "turn": 1},
]
r = score_nudge_run(log, _envelope(), case, {})
assert not r["pass"] and "stop" in r["failures"][0]
def test_allow_after_bookkeeping_permits_escalation_surfacing(self):
"""``notify`` after the bookkeeping is the escalation reaching
the operator's channel — the POINT of a legit stop, not a
violation of it. Anything outside the allowlist still fails."""
case = {"require_stop": True, "allow_after_bookkeeping": ["notify"]}
ok = [
_tasks_call("update"),
{"tool": "notify", "args": {"message": "need the token"}, "result": "", "turn": 1},
]
assert score_nudge_run(ok, _envelope(), case, {})["pass"]
stray = ok + [{"tool": "list_nodes", "args": {}, "result": "", "turn": 2}]
assert not score_nudge_run(stray, _envelope(), case, {})["pass"]
def test_no_bookkeeping_gets_its_own_label(self):
"""A run that never called ``tasks`` has no bookkeeping to have
kept working AFTER. The old fallback started the tail at index
0 and reported read-only wandering as "kept working after
bookkeeping" — a sequence that never happened, and a label that
merged two different residual classes into one bucket in the
result JSONs the residual analysis reads."""
case = {"require_stop": True}
log = [
{"tool": "list_workstreams", "args": {}, "result": "", "turn": 0},
{"tool": "list_workstreams", "args": {}, "result": "", "turn": 1},
]
r = score_nudge_run(log, _envelope(), case, {})
assert not r["pass"]
(stop,) = [f for f in r["failures"] if f.startswith("stop:")]
assert "never called tasks" in stop, stop
assert "after bookkeeping" not in stop, stop
# ...and the other class still names bookkeeping, so the two
# aggregate apart.
with_tasks = [_tasks_call("update")] + log
(other,) = [
f
for f in score_nudge_run(with_tasks, _envelope(), case, {})["failures"]
if f.startswith("stop:")
]
assert "kept working after bookkeeping" in other, other
def test_a_tasks_read_is_not_bookkeeping(self):
"""``action='list'`` records nothing. Scoring on the TOOL NAME
reported a run that listed its tasks and then dispatched work as
"kept working after bookkeeping" — naming a step that never
happened, the same describing-a-run-that-never-happened class
the never-called-tasks label was split out to remove, left open
for the read case."""
case = {"require_stop": True}
log = [
_tasks_call("list"),
{"tool": "spawn_workstream", "args": {}, "result": "", "turn": 1},
]
(stop,) = [
f for f in score_nudge_run(log, _envelope(), case, {})["failures"] if f[:5] == "stop:"
]
assert "read tasks but never recorded state" in stop, stop
assert "after bookkeeping" not in stop, stop
assert "never called tasks" not in stop, stop
def test_the_three_stop_labels_aggregate_apart(self):
"""One residual per label and no overlap: a bucket that matched
two of them would merge classes back together in the result JSON
the residual analysis reads.
The fourth log is a REJECTED write plus the stray: its envelope
was never written, so it joins the recorded-nothing bucket
rather than minting a fourth label — which is what makes the
three-way comment above the branch true for the rejected-write
case too."""
case = {"require_stop": True}
stray = {"tool": "list_nodes", "args": {}, "result": "", "turn": 9}
labels = set()
for log in (
[_tasks_call("update"), stray],
[_tasks_call("list"), stray],
[stray],
[{**_tasks_call("update"), "ok": False}, stray],
):
(stop,) = [
f
for f in score_nudge_run(log, _envelope(), case, {})["failures"]
if f[:5] == "stop:"
]
labels.add(stop.split(" (")[0])
assert len(labels) == 3, labels
def test_a_rejected_write_is_not_bookkeeping(self):
"""Effect, not intent: a schema-valid mutation whose call did
NOT land (``ok: False`` — a hallucinated task_id, an invalid
status, an over-cap note) left the envelope untouched, so it
must neither anchor the stop-rule tail nor count as recorded
state. Both directions of the old conflation:
* a TRAILING rejected write anchored the tail past every stray
before it, converting a real stop violation into a full pass
on a shipped merge-gate cell (C1's reproduction);
* a run whose ONLY tasks call was rejected was labelled "kept
working after bookkeeping" — a step that never happened — and
bucketed with the recorded-then-over-reached class.
"""
case = {"require_stop": True}
stray = {"tool": "list_nodes", "args": {}, "result": "", "turn": 1}
laundered = [
_tasks_call("update"),
stray,
{**_tasks_call("update", turn=2, task_id="tsk_hallucinated"), "ok": False},
]
r = score_nudge_run(laundered, _envelope(), case, {})
(stop,) = [f for f in r["failures"] if f.startswith("stop:")]
assert "kept working after bookkeeping" in stop, stop
assert "list_nodes" in stop, stop
only_rejected = [
{**_tasks_call("update", task_id="tsk_hallucinated"), "ok": False},
stray,
]
r2 = score_nudge_run(only_rejected, _envelope(), case, {})
(stop2,) = [f for f in r2["failures"] if f.startswith("stop:")]
assert "after bookkeeping" not in stop2, stop2
assert "read tasks but never recorded state" in stop2, stop2
def test_a_landed_write_still_anchors_with_the_flag_present(self):
"""The flag only DEMOTES: an explicit ``ok: True`` behaves like
the flagless fixtures every other test builds (the write site
always stamps it; absence means a hand-built entry)."""
case = {"require_stop": True}
stray = {"tool": "list_nodes", "args": {}, "result": "", "turn": 1}
log = [{**_tasks_call("update"), "ok": True}, stray]
(stop,) = [
f for f in score_nudge_run(log, _envelope(), case, {})["failures"] if f[:5] == "stop:"
]
assert "kept working after bookkeeping" in stop, stop
def test_a_trailing_read_does_not_launder_the_tail(self):
"""The other direction of the same conflation: anchoring on the
last ``tasks`` call of ANY kind let a closing ``action='list'``
move the anchor past every stray before it, so a run that
recorded its state, kept working, and then re-read the list
scored a clean pass."""
case = {"require_stop": True}
log = [
_tasks_call("update"),
{"tool": "list_nodes", "args": {}, "result": "", "turn": 1},
_tasks_call("list", turn=2),
]
r = score_nudge_run(log, _envelope(), case, {})
(stop,) = [f for f in r["failures"] if f.startswith("stop:")]
assert "kept working after bookkeeping" in stop, stop
assert "list_nodes" in stop, stop
def test_every_schema_action_is_classified_and_the_writes_are_bookkeeping(self):
"""The classifier is derived, not typed: the action vocabulary
comes from the tool's own schema and the read half from
production's own classifier, so this asserts the partition (a
new action classified by neither would silently join the
mutating set) and then that each mutating action really anchors
the tail."""
from turnstone.core.session import _TASKS_READ_ACTIONS, _TASKS_WRITE_ACTIONS
assert _TASKS_SCHEMA_ACTIONS == _TASKS_READ_ACTIONS | _TASKS_WRITE_ACTIONS
assert not (_TASKS_READ_ACTIONS & _TASKS_WRITE_ACTIONS)
assert _MUTATING_TASKS_ACTIONS == _TASKS_WRITE_ACTIONS
case = {"require_stop": True}
stray = {"tool": "list_nodes", "args": {}, "result": "", "turn": 1}
for action in sorted(_MUTATING_TASKS_ACTIONS):
(stop,) = [
f
for f in score_nudge_run([_tasks_call(action), stray], _envelope(), case, {})[
"failures"
]
if f.startswith("stop:")
]
assert "kept working after bookkeeping" in stop, (action, stop)
for action in sorted(_TASKS_READ_ACTIONS):
(stop,) = [
f
for f in score_nudge_run([_tasks_call(action), stray], _envelope(), case, {})[
"failures"
]
if f.startswith("stop:")
]
assert "read tasks but never recorded state" in stop, (action, stop)
def test_an_action_the_schema_does_not_declare_is_not_bookkeeping(self):
"""A hallucinated action is rejected by production, so the
envelope is untouched — counting it as bookkeeping would credit
the run with state it never recorded."""
case = {"require_stop": True}
stray = {"tool": "list_nodes", "args": {}, "result": "", "turn": 1}
for args in ({"action": "delete"}, {}, {"action": None}, None):
log = [{"tool": "tasks", "args": args, "result": "", "turn": 0}, stray]
(stop,) = [
f
for f in score_nudge_run(log, _envelope(), case, {})["failures"]
if f.startswith("stop:")
]
assert "read tasks but never recorded state" in stop, (args, stop)
def test_empty_tool_log_adds_no_stop_failure(self):
"""A run that called nothing is not a stop violation, and the
new branch must not invent one."""
r = score_nudge_run([], _envelope(), {"require_stop": True}, {})
assert r["pass"], r["failures"]
def test_no_bookkeeping_with_only_allowed_calls_adds_no_failure(self):
"""The stray filter is unchanged by the new label: a run with no
``tasks`` call whose every call is allowed still passes, whether
the allowance comes from ``allow_after_bookkeeping`` or from the
cell's own expected actions."""
allowlisted = {"require_stop": True, "allow_after_bookkeeping": ["notify"]}
log = [{"tool": "notify", "args": {"message": "need the token"}, "result": "", "turn": 0}]
assert score_nudge_run(log, _envelope(), allowlisted, {})["pass"]
expected = {
"require_stop": True,
"expect_actions": {
"mode": "contains_any",
"actions": [{"tool": "wait_for_workstream"}],
},
}
waited = [{"tool": "wait_for_workstream", "args": {}, "result": "", "turn": 0}]
assert score_nudge_run(waited, _envelope(), expected, {})["pass"]
def test_a_read_only_run_is_not_a_stop_failure_on_its_own(self):
"""``tasks`` is in the allowlist, so the read itself is never a
stray: a run that listed its tasks and then stopped adds no
failure under any of the three branches."""
case = {"require_stop": True}
assert score_nudge_run([_tasks_call("list")], _envelope(), case, {})["pass"]
assert score_nudge_run(
[_tasks_call("update"), _tasks_call("list", turn=1)], _envelope(), case, {}
)["pass"]
def test_expected_actions_contains_any(self):
case = {
"expect_actions": {
"mode": "contains_any",
"actions": [{"tool": "wait_for_workstream"}],
}
}
hit = [{"tool": "wait_for_workstream", "args": {}, "result": "", "turn": 0}]
assert score_nudge_run(hit, _envelope(), case, {})["pass"]
miss = [{"tool": "tasks", "args": {}, "result": "", "turn": 0}]
assert not score_nudge_run(miss, _envelope(), case, {})["pass"]
class TestSessionConstruction:
def test_coordinator_headless_session_constructs_with_coord_wire(self, eval_storage):
"""The passthrough seam: ``kind=COORDINATOR`` + ``coord_client``
must reach ChatSession, the wire must be the coordinator tool
list (not the CLI TOOLS constant), and the natural coordinator
prompt must compose (no override)."""
from openai import OpenAI
from turnstone.eval.nudges import CoordinatorHeadlessSession
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
session = CoordinatorHeadlessSession(
client=OpenAI(base_url="http://eval.invalid/v1", api_key="x"),
model="eval-model",
coord_client=client,
ws_id="coord-eval-1",
user_id="eval-user",
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=32768,
)
wire = {t["function"]["name"] for t in session._eval_tools}
assert "tasks" in wire and "wait_for_workstream" in wire
assert "bash" not in wire # the CLI tool set must NOT leak in
assert session.auto_approve is True
def _register_child(storage, ws_id: str, name: str, state: str) -> None:
storage.register_workstream(
ws_id,
user_id="eval-user",
name=name,
kind=WorkstreamKind.INTERACTIVE,
parent_ws_id="coord-eval-1",
state=state,
)
class TestStubClient:
def test_network_methods_are_inert(self, eval_storage):
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
out = client._post_url("http://eval.invalid/v1/api/route/x", {})
assert "error" in out and "eval environment" in out["error"]
def test_the_direct_storage_reads_are_not_overridden(self):
"""The stub's boundary is what CROSSES THE ROUTING PROXY, and it
was once drawn at the class instead: ``inspect`` and
``list_children`` — direct-storage reads production serves off
the same rows the run seeder registers — answered scripted
errors and unfiltered fixture dicts. Restoring either override
re-manufactures forbidden-rate hits on C6b (see the C6b-shaped
test below), so their absence is pinned structurally as well as
behaviourally."""
assert "inspect" not in _StubCoordinatorClient.__dict__
assert "list_children" not in _StubCoordinatorClient.__dict__
def test_c6b_recovery_reads_are_answered_not_refused(self, eval_storage):
"""The C6b shape: a stranded idle child, and a run that takes a
SCORED-CORRECT recovery (``inspect_workstream`` is in the cell's
``expect_actions``). The old class-wide override answered it
with "unavailable in the eval environment", pushing the model
toward re-running the audit — the forbidden action — so the
harness manufactured hits on its own headline metric. The real
method serves the seeded row."""
_register_child(eval_storage, "ws-child-a", "auditor", "idle")
client = _StubCoordinatorClient(
eval_storage,
coord_ws_id="coord-eval-1",
user_id="eval-user",
children=[{"ws_id": "ws-child-a", "name": "auditor", "state": "idle"}],
)
out = client.inspect("ws-child-a")
assert "error" not in out, out
assert out["state"] == "idle" and out["name"] == "auditor"
assert "messages" in out
assert "unavailable" not in json.dumps(out)
def test_inspect_misses_answer_with_the_production_shape(self, eval_storage):
"""An unknown id gets production's not-found payload — the
did-you-mean/roster recovery shape — not a stub-invented
transport error."""
_register_child(eval_storage, "ws-child-a", "auditor", "idle")
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
miss = client.inspect("deadbeef" * 4)
assert "error" in miss and "children" in miss
assert "unavailable" not in miss["error"]
def test_list_children_is_real_and_honours_the_filters(self, eval_storage):
"""Production's state filter, closed-exclusion and cross-tenant
guard all apply — the roster the model cross-checks agrees with
the seeded state and the nudge body, instead of echoing raw
fixture dicts that ignore every filter the model asked for."""
_register_child(eval_storage, "ws-child-a", "auditor", "idle")
_register_child(eval_storage, "ws-child-b", "builder", "running")
_register_child(eval_storage, "ws-child-c", "shipped", "closed")
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
default = client.list_children("coord-eval-1")
assert sorted(c["name"] for c in default["children"]) == ["auditor", "builder"]
# Production's row projection, not the fixture dict's keys.
assert default["children"][0]["parent_ws_id"] == "coord-eval-1"
assert "kind" in default["children"][0]
running = client.list_children("coord-eval-1", state="running")
assert [c["name"] for c in running["children"]] == ["builder"]
assert client.list_children("some-other-coord") == {"children": [], "truncated": False}
def test_wait_default_completes_with_stubbed_children(self, eval_storage):
client = _StubCoordinatorClient(
eval_storage,
coord_ws_id="coord-eval-1",
user_id="eval-user",
children=[{"ws_id": "ws-c1", "name": "auditor", "state": "idle"}],
)
out = client.wait_for_workstream(["ws-c1"])
assert out["complete"] is True
assert out["results"]["ws-c1"]["name"] == "auditor"
def test_wait_derives_its_answer_from_the_seeded_state(self, eval_storage):
"""The old stub returned a hardcoded terminal blob — every child
"completed its work", including ones seeded RUNNING, so the pair
arms measured a stub-invented completion. Now: a running child
is the still-running timeout shape, an idle one resolves, and
the mode semantics are production's over the real terminal set.
"""
client = _StubCoordinatorClient(
eval_storage,
coord_ws_id="coord-eval-1",
user_id="eval-user",
children=[
{"ws_id": "ws-idle", "name": "auditor", "state": "idle"},
{"ws_id": "ws-run", "name": "builder", "state": "running"},
],
)
running = client.wait_for_workstream(["ws-run"], timeout=120)
assert running["complete"] is False
assert running["results"]["ws-run"]["state"] == "running"
assert running["results"]["ws-run"]["message"] is None
assert running["elapsed"] == 120.0 # the budget a real wait would burn
idle = client.wait_for_workstream(["ws-idle"])
assert idle["complete"] is True
assert idle["results"]["ws-idle"]["state"] == "idle"
assert client.wait_for_workstream(["ws-idle", "ws-run"], mode="any")["complete"] is True
assert client.wait_for_workstream(["ws-idle", "ws-run"], mode="all")["complete"] is False
def test_wait_carries_the_seeded_childs_findings(self, eval_storage):
"""The pin the deepened world turns on: with the C6b child's
transcript seeded through the real store, the synthesized wait's
``message`` is the child's completion message — the real
``_wait_message_for`` walks the temp DB and finds it — so "the
audit finished" and "here is what it produced" arrive together,
exactly as production's wait delivers them for a finished child.
Before the fixtures deepened, this resolved ``complete`` with
the no-recent-output sentinel, and the model correctly redid
work the world claimed was done — the round-8 void."""
case = _cell("C6b_stranded_children")
child = case["children"][0]
_register_child(eval_storage, child["ws_id"], child["name"], child["state"])
_seed_child_transcripts(eval_storage, case)
client = _StubCoordinatorClient(
eval_storage,
coord_ws_id="coord-eval-1",
user_id="eval-user",
children=case["children"],
)
out = client.wait_for_workstream([child["ws_id"]])
assert out["complete"] is True
snap = out["results"][child["ws_id"]]
assert snap["message"] == child["transcript"][-1]["content"]
assert "findings" in snap["message"]
assert snap["truncated"] is False
def test_inspect_serves_the_seeded_transcript(self, eval_storage):
"""The same world through the other reader: the real ``inspect``
reconstructs the child's fixture transcript from the store the
seeder wrote — assignment then findings, in order — so C6b's
scored-correct recovery read finds the audit instead of the
``messages: []`` that voided the cell in round 8."""
case = _cell("C6b_stranded_children")
child = case["children"][0]
_register_child(eval_storage, child["ws_id"], child["name"], child["state"])
_seed_child_transcripts(eval_storage, case)
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
out = client.inspect(child["ws_id"])
assert [(m["role"], m["content"]) for m in out["messages"]] == [
(row["role"], row["content"]) for row in child["transcript"]
]
def test_wait_on_an_unregistered_id_is_the_not_found_shape(self, eval_storage):
"""Production fails a wait fast on an unobservable member, with
per-ws ``state="not_found"`` and the top-level error block. The
old stub fabricated success for ANY requested id, so a run that
waited on a hallucinated id satisfied the cell's unqualified
wait expectation."""
_register_child(eval_storage, "ws-child-a", "auditor", "idle")
client = _StubCoordinatorClient(
eval_storage,
coord_ws_id="coord-eval-1",
user_id="eval-user",
children=[{"ws_id": "ws-child-a", "name": "auditor", "state": "idle"}],
)
out = client.wait_for_workstream(["ws-nope"])
assert out["complete"] is False
assert out["results"]["ws-nope"]["state"] == "not_found"
assert out["error"] and out["not_found"]
# The hints are the real (storage-backed) recovery payload: the
# roster names the child the temp DB really carries.
assert [c["ws_id"] for c in out["children"]] == ["ws-child-a"]
def test_wait_tolerates_a_nameless_child_row(self, eval_storage):
"""A row without ``name`` is legal everywhere else — the
validator requires only ``ws_id`` and the seeder defaults the
name — so the wait synthesis must not be the one consumer that
raises on it."""
client = _StubCoordinatorClient(
eval_storage,
coord_ws_id="coord-eval-1",
user_id="eval-user",
children=[{"ws_id": "ws-anon", "state": "idle"}],
)
out = client.wait_for_workstream(["ws-anon"])
assert out["results"]["ws-anon"]["name"] == "child"
def test_scripted_stub_takes_precedence(self, eval_storage):
client = _StubCoordinatorClient(
eval_storage,
coord_ws_id="coord-eval-1",
user_id="eval-user",
tool_stubs={"wait_for_workstream": [{"complete": False, "results": {}}]},
)
assert client.wait_for_workstream(["ws-x"])["complete"] is False
def test_tasks_stay_fully_real(self, eval_storage):
"""The point of the harness: tasks execute against the temp DB
with production validation."""
client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
row = client.tasks_add("coord-eval-1", title="real", status="pending")
assert row["id"].startswith("tsk_")
# Production validation, including the unrenderable reject —
# the genuinely-invisible class, not angle brackets, which the
# operator surfaces render.
assert "error" in client.tasks_add("coord-eval-1", title=chr(0x200B) * 2)
class TestChildWorldThroughTheRunPath:
"""The C6b world end-to-end: the REAL ``_run_single_nudge`` seeds
the shipped fixture, and the collect-first moves a model would make
— wait, then inspect — are made from inside the scripted lane
against the run's own coordinator client.
This is the pin that survives a deleted call site: the validator
refuses a hollow FIXTURE, but only a run-path test can catch the
seeder call being dropped from ``_build_session`` — the fixture
would stay deep while every run's world silently hollowed back to
the round-8 shape.
"""
def test_c6b_run_world_serves_the_findings_and_scores_the_collect(self, monkeypatch):
from turnstone.eval import nudges as nudges_module
case = _cell("C6b_stranded_children")
child = case["children"][0]
seen: dict[str, Any] = {}
real_stub_cls = nudges_module._StubCoordinatorClient
def _spy_coord(*a: Any, **kw: Any) -> Any:
seen["coord_client"] = real_stub_cls(*a, **kw)
return seen["coord_client"]
class _CollectFirstLane(nudges_module.CoordinatorHeadlessSession):
def _run_headless_loop(self, **kw: Any) -> list[dict[str, Any]]:
client = seen["coord_client"]
seen["wait"] = client.wait_for_workstream([child["ws_id"]], timeout=30)
seen["inspect"] = client.inspect(child["ws_id"])
# The collect-first move, as the tool loop would log it.
return [
{
"tool": "wait_for_workstream",
"args": {"ws_ids": [child["ws_id"]]},
"result": "",
"ok": True,
"turn": 0,
}
]
monkeypatch.setattr(nudges_module, "_StubCoordinatorClient", _spy_coord)
monkeypatch.setattr(nudges_module, "CoordinatorHeadlessSession", _CollectFirstLane)
result = nudges_module._run_single_nudge(
base_url="http://eval.invalid/v1",
api_key="x",
model="eval-model",
case=case,
arm=ARM_NUDGE,
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=32768,
max_turns=2,
test_timeout=30,
verbose=False,
log_prefix="test",
)
# The wait a model issues resolves at once AND says what the
# child produced — completion and findings arrive together.
findings = child["transcript"][-1]["content"]
assert seen["wait"]["complete"] is True
assert seen["wait"]["results"][child["ws_id"]]["message"] == findings
# The inspect a model issues finds the audit inside the child.
assert [(m["role"], m["content"]) for m in seen["inspect"]["messages"]] == [
(row["role"], row["content"]) for row in child["transcript"]
]
# And the scorer files the collect-first run as the cell's pass,
# with nothing forbidden.
assert result["pass"] is True, result
assert result["forbidden"] == []
class TestToolLogEffectFlag:
"""The write site stamps ``ok`` from the executor's own error state.
``_is_mutating_tasks_call`` rules on that flag, so the stamp is the
other half of the rejected-write fix: driven END-TO-END through the
real ``_run_headless_loop`` -> ``_execute_tools`` -> ``_exec_tasks``
chain against the temp DB, with only the model lane scripted, so a
stamp that read intent (or nothing) instead of the executor's error
state fails here rather than in a sweep.
"""
@staticmethod
def _scripted_lane(monkeypatch, tool_call_turns: list[list[dict[str, Any]]]) -> None:
from turnstone.core.model_turn import ModelTurnResult
from turnstone.core.trajectory import turn_from_dict as _turn_from_dict
from turnstone.eval import core as core_module
results = [
ModelTurnResult(
turn=_turn_from_dict(
{"role": "assistant", "content": "", "tool_calls": calls}
if calls
else {"role": "assistant", "content": "done"}
),
finish_reason="tool_calls" if calls else "stop",
usage=None,
tool_calls=calls,
)
for calls in tool_call_turns
]
sequence = iter(results)
monkeypatch.setattr(core_module, "model_turn", lambda *a, **k: next(sequence))
@staticmethod
def _wire_call(call_id: str, name: str, args: dict[str, Any]) -> dict[str, Any]:
return {
"id": call_id,
"type": "function",
"function": {"name": name, "arguments": json.dumps(args)},
}
def test_a_rejected_tasks_write_is_stamped_not_ok(self, eval_storage, monkeypatch):
from openai import OpenAI
from turnstone.eval.nudges import CoordinatorHeadlessSession, _is_mutating_tasks_call
coord_client = _StubCoordinatorClient(
eval_storage, coord_ws_id="coord-eval-1", user_id="eval-user"
)
seeded = coord_client.tasks_add("coord-eval-1", title="real row", status="pending")
session = CoordinatorHeadlessSession(
client=OpenAI(base_url="http://eval.invalid/v1", api_key="x"),
model="eval-model",
coord_client=coord_client,
ws_id="coord-eval-1",
user_id="eval-user",
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=32768,
)
self._scripted_lane(
monkeypatch,
[
[
self._wire_call(
"call_landed",
"tasks",
{"action": "update", "task_id": seeded["id"], "status": "done"},
),
self._wire_call(
"call_rejected",
"tasks",
{"action": "update", "task_id": "tsk_hallucinated", "status": "done"},
),
],
[],
],
)
from turnstone.core.trajectory import Turn
session.messages.append(Turn.user("do the bookkeeping"))
session._msg_tokens.append(1)
try:
log = session._run_headless_loop(max_turns=3)
finally:
session.close()
assert [entry["tool"] for entry in log] == ["tasks", "tasks"]
assert [entry["ok"] for entry in log] == [True, False]
# The flag is what the bookkeeping classifier rules on.
assert _is_mutating_tasks_call(log[0])
assert not _is_mutating_tasks_call(log[1])
# And the landed write really landed while the rejected one
# really did not — the flag mirrors the envelope, not the wish.
envelope = coord_client.tasks_get("coord-eval-1")
assert [t["status"] for t in envelope["tasks"]] == ["done"]
class TestRunResourceLifecycle:
"""A run must leave nothing behind — a sweep is hundreds of runs,
and every leaked connection pool is an fd that a later run cannot
open. Exhaustion surfaces as
``failures: ["harness: …"]``: a plausible red 0% attributed to the
body under test, the same dishonesty class the canary probe and the
sweep-start validator exist to prevent.
Each test drives the REAL ``_run_single_nudge`` with only the model
lane stubbed, and asserts on the artifact (a closed transport, an
absent directory, the process cwd) rather than on a recorded
intention to close.
"""
_CASE = {"id": "L_lifecycle", "arms": [ARM_NUDGE], "tasks": [{"title": "audit the config"}]}
@staticmethod
def _drive(
monkeypatch,
*,
close_raises: type[BaseException] | None = None,
init_raises: bool = False,
teardown_reset_raises: bool = False,
cwd_restore_raises: bool = False,
loop_behavior: str | None = None,
poison: str | None = None,
fast_retries: bool = False,
test_timeout: int = 30,
):
"""One real run with the generation lane stubbed out.
Returns the objects the run built, keyed for assertion. The
session subclass replaces ONLY ``_run_headless_loop`` (the sole
step that would reach a model), so construction, seeding, the
retry loop, the wall clock and the whole teardown path are
production code.
*loop_behavior*: ``"raise_once"`` / ``"raise_always"`` model a
transient / persistent generation failure; ``"hang"`` parks the
worker on the session's own cancel event (so the wall clock
fires and the worker still exits promptly once cancelled).
*poison* corrupts the post-run ground truth from INSIDE the run,
against the current attempt's storage: ``"read_raises"`` makes
the raw config read blow up, ``"corrupt"`` writes an unparseable
envelope, ``"missing"`` erases the seeded key. *fast_retries*
neutralizes the retry backoff sleeps.
"""
from turnstone.eval import nudges as nudges_module
made: dict[str, Any] = {}
if cwd_restore_raises:
launch_cwd = os.getcwd()
real_chdir = os.chdir
budget = [1]
def _chdir_failing_restore(path: str) -> None:
# Models the launch directory going away mid-sweep (it
# is removed, or its mount drops): the chdir INTO the
# workdir still works, the restore does not. One-shot,
# so the test can put the process back afterwards.
if os.path.abspath(path) == launch_cwd and budget:
budget.pop()
raise OSError(2, "No such file or directory", path)
real_chdir(path)
monkeypatch.setattr(os, "chdir", _chdir_failing_restore)
if fast_retries:
monkeypatch.setattr(time, "sleep", lambda _s: None)
real_mkdtemp = tempfile.mkdtemp
def _spy_mkdtemp(*a: Any, **kw: Any) -> str:
path = real_mkdtemp(*a, **kw)
if kw.get("prefix") == "turnstone_eval_nudge_":
made["workdir"] = path
return path
real_openai = nudges_module.OpenAI
def _spy_openai(**kw: Any) -> Any:
made["run_client"] = client = real_openai(**kw)
made.setdefault("run_clients", []).append(client)
return client
real_stub_cls = nudges_module._StubCoordinatorClient
def _spy_coord(*a: Any, **kw: Any) -> Any:
made["coord_client"] = client = real_stub_cls(*a, **kw)
return client
class _StubbedLaneSession(nudges_module.CoordinatorHeadlessSession):
def __init__(self, **kw: Any) -> None:
super().__init__(**kw)
made["session"] = self
made.setdefault("sessions", []).append(self)
def _run_headless_loop(self, **kw: Any) -> list[dict[str, Any]]:
made["cwd_in_run"] = os.getcwd()
made["loop_calls"] = made.get("loop_calls", 0) + 1
# What THIS attempt's world contains — a retried attempt
# must see a freshly seeded envelope, not the previous
# attempt's rows plus a re-seed.
made.setdefault("seeded_counts", []).append(
len(made["coord_client"].tasks_get("coord-eval-1")["tasks"])
)
if poison == "read_raises":
storage = get_storage()
def _read_boom(*a: Any, **k: Any) -> dict[str, Any]:
raise RuntimeError("storage read blew up")
storage.load_workstream_config = _read_boom # type: ignore[method-assign]
elif poison == "corrupt":
get_storage().save_workstream_config("coord-eval-1", {"tasks": "{not json"})
elif poison == "missing":
get_storage().save_workstream_config("coord-eval-1", {"tasks": ""})
if loop_behavior == "hang":
self._cancelled.wait(10)
return []
if loop_behavior == "raise_always" or (
loop_behavior == "raise_once" and made["loop_calls"] == 1
):
raise RuntimeError("transient boom: connection reset by peer")
return []
def close(self) -> None:
if close_raises is not None:
# Models a raise from one of ``ChatSession.close()``'s
# UNGUARDED steps (only the ``_coord_client`` step is
# exception-guarded), so nothing after it in the real
# close runs either — including that client's close.
raise close_raises("teardown blew up mid-close")
super().close()
monkeypatch.setattr(tempfile, "mkdtemp", _spy_mkdtemp)
monkeypatch.setattr(nudges_module, "OpenAI", _spy_openai)
monkeypatch.setattr(nudges_module, "_StubCoordinatorClient", _spy_coord)
monkeypatch.setattr(nudges_module, "CoordinatorHeadlessSession", _StubbedLaneSession)
if init_raises:
def _boom(*a: Any, **kw: Any) -> None:
raise RuntimeError("storage init failed")
monkeypatch.setattr(nudges_module, "init_storage", _boom)
if teardown_reset_raises:
real_reset = nudges_module.reset_storage
resets: list[int] = []
def _reset_then_fail(*a: Any, **kw: Any) -> None:
# The run does its setup reset first; only the ``finally``
# one fails, so the failure lands on a run that otherwise
# completed.
resets.append(1)
if len(resets) > 1:
raise RuntimeError("storage reset failed")
real_reset(*a, **kw)
monkeypatch.setattr(nudges_module, "reset_storage", _reset_then_fail)
try:
made["result"] = nudges_module._run_single_nudge(
base_url="http://eval.invalid/v1",
api_key="x",
model="eval-model",
case=TestRunResourceLifecycle._CASE,
arm=ARM_NUDGE,
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=32768,
max_turns=2,
test_timeout=test_timeout,
verbose=False,
log_prefix="test",
)
except BaseException as exc: # noqa: BLE001 - the raise IS the fixture
made["raised"] = exc
return made
def test_run_single_nudge_closes_its_clients(self, monkeypatch):
"""Both per-run HTTP owners are released. Asserted on the
transports themselves — a spy counting ``close()`` calls would
pass against a session that swallowed the call."""
made = self._drive(monkeypatch)
assert "raised" not in made, made.get("raised")
assert made["result"]["pass"] is True, made["result"]
assert made["run_client"].is_closed(), "the per-run OpenAI client leaked its pool"
assert made["coord_client"]._http.is_closed, "the stub coordinator client leaked its pool"
# The session's own close ran to completion: the background-shell
# registry is its last-but-one step and latches closed.
assert made["session"]._background_shells._closed is True
def test_init_storage_failure_still_removes_the_workdir(self, monkeypatch):
"""Storage setup lives INSIDE the ``try`` whose ``finally`` owns
the directory. Before that move a failing ``init_storage``
stranded one temp dir per attempt, permanently."""
cwd_before = os.getcwd()
made = self._drive(monkeypatch, init_raises=True)
assert isinstance(made["raised"], RuntimeError)
assert "workdir" in made, "the run never got as far as making one"
assert not os.path.exists(made["workdir"])
assert os.getcwd() == cwd_before
def test_a_failing_storage_reset_still_removes_the_workdir(self, monkeypatch):
"""The teardown reset is nested, not suppressed.
A storage fault is real and must surface — so the raise
propagates and the sweep records the run as a ``harness:``
failure — but it must not take the ``rmtree`` with it.
Flattened, this is the SETUP-path defect reappearing on the
teardown path: one directory stranded permanently per run, i.e.
the defect inside the fix for it.
"""
cwd_before = os.getcwd()
made = self._drive(monkeypatch, teardown_reset_raises=True)
assert isinstance(made["raised"], RuntimeError) # the fault surfaced
assert "result" not in made # a finally-raise discards the return
assert not os.path.exists(made["workdir"]) # and cost the run nothing
assert os.getcwd() == cwd_before
reset_storage() # the patched reset never got to do its job
def test_a_raising_session_close_still_completes_the_cleanup(self, monkeypatch):
"""A failing teardown costs the run NOTHING else.
Each close is suppressed on its own, so a raise inside
``ChatSession.close()`` cannot take the closes after it, the
storage reset or the temp dir with it — and cannot even reach
the sweep as a ``harness:`` failure, since the run's result is
already computed by then.
"""
cwd_before = os.getcwd()
made = self._drive(monkeypatch, close_raises=RuntimeError)
assert "raised" not in made, made.get("raised") # contained, not propagated
assert made["result"]["pass"] is True
assert made["cwd_in_run"] == made["workdir"] # it really did chdir in
assert os.getcwd() == cwd_before
assert not is_storage_initialized()
assert not os.path.exists(made["workdir"])
# The closes after the raising one still ran — including the
# coordinator client that the failed ``session.close()`` never
# reached, which is why that close is kept as its own step.
assert made["run_client"].is_closed()
assert made["coord_client"]._http.is_closed
def test_a_ctrl_c_during_close_still_restores_the_cwd(self, monkeypatch):
"""What the cwd-restore-FIRST ordering is worth, second case.
A suppressed close cannot abort the block at all, so the
ordering is not what saves an ordinary teardown failure. It
earns its keep on the two raises that DO leave the block early:
the unsuppressed storage reset (the test above) and a
``BaseException``, which no suppression catches — realistically
a Ctrl-C landing in ``_background_shells.close()``'s bounded
join, the one blocking window a long sweep offers an impatient
operator. The residues are not equal: a leaked temp dir is
inert and visible, while a process left chdir'd into a deleted
directory silently breaks every subsequent run — which this
test's own process would then demonstrate by dying at its next
``os.getcwd()``. The restore goes first so an interrupt can
only cost the inert one.
"""
cwd_before = os.getcwd()
made = self._drive(monkeypatch, close_raises=KeyboardInterrupt)
assert isinstance(made["raised"], KeyboardInterrupt)
assert made["cwd_in_run"] == made["workdir"]
assert os.getcwd() == cwd_before # survived the abort — it ran first
# The honest cost of an abort mid-block, asserted rather than
# implied: everything after the interrupt is skipped.
assert os.path.exists(made["workdir"])
assert is_storage_initialized()
reset_storage()
made["run_client"].close()
made["coord_client"].close()
shutil.rmtree(made["workdir"], ignore_errors=True)
def test_a_failing_cwd_restore_still_completes_the_cleanup(self, monkeypatch):
"""The teardown's FIRST statement is guarded too.
The restore is deliberately first, and every close after it is
suppressed on its own — but an unguarded ``os.chdir`` raise (the
launch directory removed, or its mount dropped mid-sweep) would
skip all three closes, the storage reset AND the rmtree,
reinstating the per-run leak the block exists to prevent, in the
one situation where the operator can least afford it. The
block's own comment claims nothing in it can cost the rest; that
claim must not be falsifiable by the line that implements it.
The failure is logged, not swallowed silently, and it is NOT
re-raised: this run's result is already computed and is still
honest. The cost that remains is asserted below rather than
implied — the process is left inside a directory the rmtree then
removes, so the NEXT run fails early and visibly.
"""
cwd_before = os.getcwd()
made = self._drive(monkeypatch, cwd_restore_raises=True)
os.chdir(cwd_before) # first: the process is standing in a deleted dir
assert "raised" not in made, made.get("raised") # contained, not propagated
assert made["result"]["pass"] is True # and the measurement survived
assert made["cwd_in_run"] == made["workdir"] # the restore really was the failing call
# Everything the unguarded chdir used to skip:
assert made["run_client"].is_closed(), "the per-run OpenAI client leaked its pool"
assert made["coord_client"]._http.is_closed, "the stub coordinator client leaked its pool"
assert made["session"]._background_shells._closed is True
assert not is_storage_initialized()
assert not os.path.exists(made["workdir"])
def test_a_failing_storage_reset_still_surfaces(self, monkeypatch):
"""The property the chdir guard must not have bought at the cost
of: a storage fault is real and still propagates. Guarding the
restore with a suppression wide enough to swallow the reset —
or wrapping the whole block — would take this with it.
"""
cwd_before = os.getcwd()
made = self._drive(monkeypatch, teardown_reset_raises=True)
assert isinstance(made["raised"], RuntimeError)
assert "storage reset failed" in str(made["raised"])
assert os.getcwd() == cwd_before
reset_storage()
def test_a_transient_generation_failure_is_retried_with_fresh_state(self, monkeypatch):
"""The mid-sweep hiccup the canary cannot see.
The probe samples only sweep start and sweep end, so one
connection reset on run 4 of 10 used to file 90% as a body
regression. The shared lifecycle retries the attempt — and
because scoring is STATE-FIRST, the retry rebuilds the whole
seeded world: an attempt that inherited the failed attempt's
envelope would score expectations the model never earned.
"""
made = self._drive(monkeypatch, loop_behavior="raise_once", fast_retries=True)
assert "raised" not in made, made.get("raised")
assert made["result"]["pass"] is True
assert made["loop_calls"] == 2 # failed once, retried once
# Each attempt saw a FRESHLY seeded envelope: exactly the cell's
# one row, not the previous attempt's row plus a re-seed.
assert made["seeded_counts"] == [1, 1]
# The failed attempt's session was closed before being replaced
# — the worker had already raised, so the drop-unclosed rule is
# the timeout path's alone.
assert all(s._background_shells._closed for s in made["sessions"])
assert all(c.is_closed() for c in made["run_clients"])
def test_every_failed_attempts_session_is_closed(self, monkeypatch):
"""A persistent failure exhausts the 3 attempts and surfaces as
a harness failure — and every attempt's fully-built session is
CLOSED on the way, not abandoned with its listener
registrations and background-shell registry."""
made = self._drive(monkeypatch, loop_behavior="raise_always", fast_retries=True)
assert isinstance(made["raised"], RuntimeError)
assert "transient boom" in str(made["raised"])
assert made["loop_calls"] == 3 # the 3x retry really ran
assert len(made["sessions"]) == 3
assert all(s._background_shells._closed for s in made["sessions"])
assert all(c.is_closed() for c in made["run_clients"])
assert not os.path.exists(made["workdir"])
assert not is_storage_initialized()
def test_a_hung_generation_is_bounded_by_the_wall_clock(self, monkeypatch):
"""The per-request HTTP transport timeout cannot bound a STREAM — a
trickling response resets the read timeout indefinitely — so
without the executor wall clock a hung generation occupies a run
slot forever and is scored as a body regression when the sweep
is finally killed. The wall clock converts it to a loud
``harness:`` timeout; the hung attempt's session is DROPPED
unclosed, deliberately (the worker is still inside the drive;
close would trade a bounded leak for a blocked teardown).
"""
cwd_before = os.getcwd()
made = self._drive(monkeypatch, loop_behavior="hang", test_timeout=1)
assert isinstance(made["raised"], TimeoutError)
assert made["loop_calls"] == 1 # a timeout aborts, never retries
assert made["sessions"][0]._background_shells._closed is False # dropped
assert made["run_clients"][0].is_closed() # the transport IS closed
assert os.getcwd() == cwd_before
assert not os.path.exists(made["workdir"])
assert not is_storage_initialized()
# Test hygiene, not production: reap the deliberately-dropped
# session once its cancelled worker has exited.
made["sessions"][0].close()
def test_a_post_run_storage_fault_is_a_harness_failure(self, monkeypatch):
"""The post-run envelope read is scoring GROUND TRUTH, and it
was the module's only read whose failure was attributed to the
body under test: ``load_task_envelope`` fails open, so a
transient storage fault came back as an empty envelope and
scored as the model having deleted its tasks. The probe read
raises instead, landing in the sweep's ``harness:`` bucket."""
made = self._drive(monkeypatch, poison="read_raises", fast_retries=True)
assert "result" not in made # never scored as a model result
assert isinstance(made["raised"], RuntimeError)
assert "storage read blew up" in str(made["raised"])
def test_a_corrupt_post_run_envelope_refuses_to_score(self, monkeypatch):
"""The corrupt flag is honoured, not discarded: corruption used
to read as emptiness, byte-identical in the result JSON to a
coordinator that deleted the task it was told to escalate."""
made = self._drive(monkeypatch, poison="corrupt", fast_retries=True)
assert "result" not in made
assert isinstance(made["raised"], RuntimeError)
assert "corrupt" in str(made["raised"])
def test_a_vanished_post_run_envelope_refuses_to_score(self, monkeypatch):
"""The seed WROTE the tasks key; a post-run read without it is a
storage fault, not a model that emptied its list (no production
path deletes the key)."""
made = self._drive(monkeypatch, poison="missing", fast_retries=True)
assert "result" not in made
assert isinstance(made["raised"], RuntimeError)
assert "missing from storage" in str(made["raised"])
class TestHarnessFailureRecord:
"""A run the harness lost must file a record the readers can sum.
The failure record is shape-uniform with a success record — same
keys, zero-valued ``elapsed`` / ``usage`` — so a consumer summing
cost or wall-clock across a cell's ``runs`` never meets a missing
key on exactly the sweeps whose totals it most wants to read.
"""
def test_the_failure_record_carries_the_success_records_keys(self, monkeypatch):
from turnstone.eval import nudges as nudges_module
calls = {"n": 0}
def _flaky_run(**kw: Any) -> dict[str, Any]:
calls["n"] += 1
if calls["n"] == 1:
return {
"pass": True,
"failures": [],
"forbidden": [],
"actions": ["tasks"],
"elapsed": 1.25,
"usage": {"prompt": 10, "completion": 5},
}
raise RuntimeError("endpoint hiccup")
monkeypatch.setattr(nudges_module, "tool_call_canary", lambda *a, **k: True)
monkeypatch.setattr(nudges_module, "_run_single_nudge", _flaky_run)
out = run_nudge_response(
base_url="http://eval.invalid/v1",
api_key="x",
model="eval-model",
cells=[{"id": "X_shape", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]}],
n_runs=2,
)
success, failure = out["cells"]["X_shape"][ARM_NUDGE]["runs"]
assert set(failure) == set(success)
assert failure["failures"] == ["harness: endpoint hiccup"]
assert failure["elapsed"] == 0
assert failure["usage"] == {"prompt": 0, "completion": 0}
class TestCanaryFloor:
"""The probe's budget is floored, never inherited raw.
The canary's own docstring records the starved-probe trap (a
thinking model burns a small budget inside its reasoning block and
the probe reads as a dead parser), and the runner used to re-enable
it by passing the sweep's ``max_tokens`` with no floor — an operator
smoke-sweeping with ``--max-tokens 512`` got an ABORT telling them
to restart a healthy container.
"""
@staticmethod
def _probe_budgets(monkeypatch, sweep_max_tokens: int) -> list[int]:
from turnstone.eval import nudges as nudges_module
seen: list[int] = []
def _spy_canary(*a: Any, **k: Any) -> bool:
seen.append(k["max_tokens"])
return True
monkeypatch.setattr(nudges_module, "tool_call_canary", _spy_canary)
monkeypatch.setattr(
nudges_module,
"_run_single_nudge",
lambda **kw: {
"pass": True,
"failures": [],
"forbidden": [],
"actions": [],
"elapsed": 0,
"usage": {"prompt": 0, "completion": 0},
},
)
run_nudge_response(
base_url="http://eval.invalid/v1",
api_key="x",
model="eval-model",
cells=[{"id": "X_floor", "arms": [ARM_NUDGE], "tasks": [_OPEN_TASK]}],
n_runs=1,
max_tokens=sweep_max_tokens,
)
return seen
def test_a_starved_sweep_budget_cannot_reach_the_probe(self, monkeypatch):
# Three probes, not two: the stub returns zero tool calls on a
# nudge-class arm, so the mid-sweep tripwire fires its own
# canary between the bracketing pair — and it must honour the
# same floor, or the tripwire re-opens the starved-probe trap.
assert self._probe_budgets(monkeypatch, 512) == [8192, 8192, 8192]
def test_a_generous_sweep_budget_widens_the_probe(self, monkeypatch):
assert self._probe_budgets(monkeypatch, 50000) == [50000, 50000, 50000]
class TestCliNRuns:
"""``--n-runs 0`` is the validate-and-canary dry run — both run
before any generation — and the ``or``-resolution silently replaced
it with the full default grid against a personally-funded endpoint.
The file's two sibling resolution sites already used ``is not
None``; this pins the nudge path to the same rule.
"""
@staticmethod
def _resolved_n_runs(monkeypatch, tmp_path, n_runs_arg: int | None) -> int:
import argparse
from turnstone.eval import cli as cli_module
from turnstone.eval import nudges as nudges_module
seen: dict[str, Any] = {}
def _fake_sweep(**kw: Any) -> dict[str, Any]:
seen["n_runs"] = kw["n_runs"]
return {"model": kw["model"], "cells": {}}
monkeypatch.setattr(nudges_module, "run_nudge_response", _fake_sweep)
args = argparse.Namespace(
cells=None,
body_override=None,
base_url="http://eval.invalid/v1",
n_runs=n_runs_arg,
temperature=0.7,
max_tokens=1024,
reasoning_effort="medium",
context_window=32768,
test_timeout=30,
verbose=False,
output=str(tmp_path / "out.json"),
)
cli_module._run_nudges_cli(args, "eval-model", "key")
return seen["n_runs"]
def test_an_explicit_zero_is_passed_through(self, monkeypatch, tmp_path):
assert self._resolved_n_runs(monkeypatch, tmp_path, 0) == 0
def test_an_omitted_flag_still_defaults_to_ten(self, monkeypatch, tmp_path):
assert self._resolved_n_runs(monkeypatch, tmp_path, None) == 10
class TestMidSweepTripwire:
"""The dead-parser tripwire: quiet nudge-class arms re-fire the
canary; a failed probe aborts the sweep, a healthy one stamps the
anomaly into the fingerprint. Born from the 2026-07-28 outage,
where 144 of 240 runs returned zero tool calls between two green
bracketing canaries."""
@staticmethod
def _cell_out(empty_runs: int, total: int = 10, arm: str = ARM_NUDGE):
runs = [
{
"pass": False,
"failures": [],
"forbidden": [],
"actions": [],
"elapsed": 0,
"usage": {},
}
for _ in range(empty_runs)
] + [
{
"pass": True,
"failures": [],
"forbidden": [],
"actions": ["tasks"],
"elapsed": 1,
"usage": {},
}
for _ in range(total - empty_runs)
]
return {arm: {"n": total, "pass_rate": 0.0, "forbidden_rate": 0.0, "runs": runs}}
def test_dead_parser_aborts_the_sweep(self, monkeypatch):
import turnstone.eval.nudges as nudges_mod
monkeypatch.setattr(nudges_mod, "tool_call_canary", lambda *a, **k: False)
out = {"body": {"cells": {}}}
with pytest.raises(SystemExit, match="died mid-sweep"):
nudges_mod._tripwire_check(
"C4_finished_unmarked",
self._cell_out(empty_runs=10),
out,
base_url="http://x/v1",
api_key="k",
model="m",
max_tokens=8192,
)
def test_quiet_arms_with_healthy_canary_are_stamped_not_fatal(self, monkeypatch):
import turnstone.eval.nudges as nudges_mod
probes = []
monkeypatch.setattr(
nudges_mod, "tool_call_canary", lambda *a, **k: probes.append(1) or True
)
out = {"body": {"cells": {}}}
nudges_mod._tripwire_check(
"C4_finished_unmarked",
self._cell_out(empty_runs=6),
out,
base_url="http://x/v1",
api_key="k",
model="m",
max_tokens=8192,
)
assert probes, "the canary must be re-fired on a quiet nudge-class arm"
assert out["body"]["cells"]["C4_finished_unmarked"]["quiet_nudge_arms"] == [ARM_NUDGE]
def test_bare_continue_silence_is_legitimate_and_never_probes(self, monkeypatch):
import turnstone.eval.nudges as nudges_mod
monkeypatch.setattr(
nudges_mod,
"tool_call_canary",
lambda *a, **k: (_ for _ in ()).throw(AssertionError("probed")),
)
out = {"body": {"cells": {}}}
nudges_mod._tripwire_check(
"C4_finished_unmarked",
self._cell_out(empty_runs=10, arm=ARM_BARE_CONTINUE),
out,
base_url="http://x/v1",
api_key="k",
model="m",
max_tokens=8192,
)
assert out["body"]["cells"] == {}
def test_minority_silence_does_not_trip(self, monkeypatch):
import turnstone.eval.nudges as nudges_mod
monkeypatch.setattr(
nudges_mod,
"tool_call_canary",
lambda *a, **k: (_ for _ in ()).throw(AssertionError("probed")),
)
out = {"body": {"cells": {}}}
nudges_mod._tripwire_check(
"C1_legit_stop_approval",
self._cell_out(empty_runs=5),
out,
base_url="http://x/v1",
api_key="k",
model="m",
max_tokens=8192,
)
assert out["body"]["cells"] == {}
Reference in New Issue View Git Blame Copy Permalink