dongarandClaude Opus 5
2b7651fb37
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m7s
Completes step 7. - expert.py: one stateless call for a question needing no repo access. Structured output is contracted at the prompt level, since vendors differ on JSON enforcement, so the reply is parsed and validated locally with one retry — tolerating a fenced code block, because models wrap JSON habitually, but not missing fields - The per-task budget is counted from the canonical log rather than memory, so it survives a restart and cannot hand a circling executor a fresh allowance. Exhaustion raises rather than answering: a caller out of consults should get on with the work or escalate - Consults are recorded as delegation_started/completed with role consultant. A16 asks for a "consult/delegation eval set", and reusing the composite events gets consult evaluation for free rather than inventing a parallel record - evals.py pairs starts with completions into eval records, with verdicts kept outside the canonical log: the log records what happened, a verdict is a later judgement about it, and judgements get revised. Schema v3 adds the outcomes table, so a v1 store now traverses two migrations — a better exercise of the hook than one - ask_expert is offered to the executor as a tool only when a desk is configured, in addition to the observable escalation triggers and never instead of them (A12). Consulting is not mutating, so a completion claim backed only by consults is still challenged Found live: passing an empty message list and putting the question in the system prompt fails outright — the Messages API requires at least one non-system message. The question now travels as a synthesised user message that is never appended to the log, which keeps the consult stateless while remaining a valid request. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
"""A real consult over the normalized route (A14).
|
|
|
|
The point is the prompt-level contract: a live model must produce something
|
|
that parses against our local schema, since no API-level enforcement exists.
|
|
"""
|
|
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nisse_dev.envfile import load_env_file
|
|
from nisse_dev.evals import collect
|
|
from nisse_dev.expert import BudgetExhausted, ExpertDesk
|
|
from nisse_dev.llm import LiteLLMClient, RoleMap
|
|
from nisse_dev.oplog import OpLog
|
|
from nisse_dev.state import StateStore
|
|
|
|
load_env_file(Path(__file__).resolve().parents[2] / ".env")
|
|
|
|
SENTINEL = "NISSE_LIVE_PLANNER_MODEL"
|
|
|
|
pytestmark = [
|
|
pytest.mark.integration,
|
|
pytest.mark.skipif(
|
|
not os.environ.get(SENTINEL),
|
|
reason=f"no live frontier route configured ({SENTINEL} unset)",
|
|
),
|
|
]
|
|
|
|
|
|
def desk(tmp_path: Path, budget: int = 2) -> tuple[ExpertDesk, StateStore]:
|
|
model = os.environ[SENTINEL]
|
|
roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
|
|
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
|
|
return ExpertDesk(LiteLLMClient.from_env(), store, roles, budget=budget), store
|
|
|
|
|
|
def test_live_consult_parses_against_the_local_schema(tmp_path: Path) -> None:
|
|
expert, store = desk(tmp_path)
|
|
|
|
answer = expert.ask(
|
|
"For a single-process agent storing a few thousand rows of task state, "
|
|
"is SQLite sufficient, or is PostgreSQL warranted? Answer briefly."
|
|
)
|
|
|
|
assert answer.confidence in {"high", "medium", "low"}
|
|
assert len(answer.answer) > 20
|
|
# The consult is captured for A16 without any extra bookkeeping.
|
|
[record] = collect(store)
|
|
assert record.role == "consultant"
|
|
assert record.exit_status == "answered"
|
|
|
|
|
|
def test_live_budget_stops_a_circling_caller(tmp_path: Path) -> None:
|
|
expert, _ = desk(tmp_path, budget=1)
|
|
expert.ask("Name one advantage of WAL mode in SQLite. Be brief.")
|
|
with pytest.raises(BudgetExhausted):
|
|
expert.ask("And another?")
|