javadandCursor
b93ecf1691
Keep source, config, and deploy files. Leave runtime data, uploads, and logs out of git. Co-authored-by: Cursor <cursoragent@cursor.com>
115 lines
4.8 KiB
Python
115 lines
4.8 KiB
Python
"""Budget, evidence, and user-mode reasoning for the 10 production skills."""
|
|
|
|
import json
|
|
|
|
from kamidso.contracts.chat import KamidsoChatRequest
|
|
from kamidso.engine.app.openapi.api_v1.tools.skill_runtime import (
|
|
blocked_observation,
|
|
budget_block,
|
|
call_fingerprint,
|
|
compact_handoff_evidence,
|
|
empty_skill_execution,
|
|
parse_skill_contract,
|
|
record_skill_result,
|
|
skill_minimum_for,
|
|
)
|
|
from kamidso.engine.core.reasoning_policy import (
|
|
ProviderReasoningAdapter,
|
|
ReasoningLevel,
|
|
ReasoningPolicyResolver,
|
|
ReasoningUnavailableError,
|
|
UserMode,
|
|
bind_turn,
|
|
current_effective_level,
|
|
reset_turn,
|
|
set_active_skill,
|
|
)
|
|
|
|
|
|
def test_fast_deep_and_toolbar_modes():
|
|
resolver = ReasoningPolicyResolver()
|
|
assert resolver.resolve(UserMode.FAST) == ReasoningLevel.OFF
|
|
assert resolver.resolve(UserMode.MEDIUM) == ReasoningLevel.MEDIUM
|
|
assert resolver.resolve(UserMode.DEEP) == ReasoningLevel.HIGH
|
|
assert resolver.resolve(UserMode.FAST, ReasoningLevel.HIGH) == ReasoningLevel.OFF
|
|
assert resolver.resolve(UserMode.MEDIUM, ReasoningLevel.HIGH) == ReasoningLevel.MEDIUM
|
|
assert resolver.resolve(UserMode.DEEP, ReasoningLevel.LOW) == ReasoningLevel.HIGH
|
|
|
|
|
|
def test_chain_keeps_toolbar_mode_when_skills_activate():
|
|
token = bind_turn("fast")
|
|
try:
|
|
set_active_skill("profitability-cost-intelligence", ReasoningLevel.MEDIUM)
|
|
assert current_effective_level() == ReasoningLevel.OFF
|
|
set_active_skill("root-cause-investigator", ReasoningLevel.HIGH)
|
|
assert current_effective_level() == ReasoningLevel.OFF
|
|
set_active_skill(None, None)
|
|
assert current_effective_level() == ReasoningLevel.OFF
|
|
finally:
|
|
reset_turn(token)
|
|
|
|
|
|
def test_provider_fallback_or_clear_error():
|
|
adapter = ProviderReasoningAdapter(supported_efforts=frozenset({"none", "low"}))
|
|
try:
|
|
adapter.effort_for(ReasoningLevel.HIGH, required=True)
|
|
raise AssertionError("expected ReasoningUnavailableError")
|
|
except ReasoningUnavailableError:
|
|
pass
|
|
effort, fallback = adapter.effort_for(ReasoningLevel.HIGH, fallback_model="reasoner-high")
|
|
assert fallback == "reasoner-high"
|
|
assert effort == "high"
|
|
|
|
|
|
def test_chat_request_carries_reasoning_mode():
|
|
assert KamidsoChatRequest.from_body({"message": "hi"}).reasoning_mode == "fast"
|
|
assert KamidsoChatRequest.from_body({"message": "hi", "reasoning_mode": "medium"}).reasoning_mode == "medium"
|
|
assert KamidsoChatRequest.from_body({"message": "hi", "reasoning_mode": "deep"}).reasoning_mode == "deep"
|
|
|
|
|
|
def test_chat_request_carries_report_mode():
|
|
assert KamidsoChatRequest.from_body({"message": "hi"}).report_mode is False
|
|
assert KamidsoChatRequest.from_body({"message": "hi", "report_mode": True}).report_mode is True
|
|
assert KamidsoChatRequest.from_body({"message": "hi", "reportMode": "on"}).report_mode is True
|
|
assert KamidsoChatRequest.from_body({"message": "hi", "report_mode": "false"}).report_mode is False
|
|
|
|
|
|
def test_skill_budget_and_shared_evidence():
|
|
state = {"skill_execution": empty_skill_execution()}
|
|
args = {"input_file": "a.csv"}
|
|
record_skill_result(
|
|
state,
|
|
"profitability-cost-intelligence",
|
|
args,
|
|
{
|
|
"skill_id": "profitability-cost-intelligence",
|
|
"status": "completed",
|
|
"summary": "margin dropped",
|
|
"evidence": [{"claim": "Gross margin declined", "metric": "gross_margin"}],
|
|
},
|
|
)
|
|
assert budget_block(state, "profitability-cost-intelligence", args)["status"] == "duplicate_call"
|
|
evidence = compact_handoff_evidence(state)
|
|
assert budget_block(state, "root-cause-investigator", {"evidence": evidence}) is None
|
|
payload = json.loads(blocked_observation(budget_block(state, "profitability-cost-intelligence", args)))
|
|
assert "duplicate_call" in payload["chunks"][0]["content"]
|
|
|
|
|
|
def test_max_hops_and_fingerprint_ignore_previous_evidence():
|
|
state = {"skill_execution": empty_skill_execution()}
|
|
for index in range(6):
|
|
record_skill_result(state, f"skill-{index}", {"n": index}, {"status": "completed"})
|
|
assert budget_block(state, "sales-intelligence", {"n": 99})["reason"] == "max_skill_hops"
|
|
assert call_fingerprint("sales-intelligence", {"input_file": "a.csv", "previous_evidence": [1]}) == call_fingerprint(
|
|
"sales-intelligence", {"input_file": "a.csv"}
|
|
)
|
|
|
|
|
|
def test_parse_skill_contract_and_catalog_minima():
|
|
body = {"skill_id": "sales-intelligence", "status": "completed", "summary": "done"}
|
|
raw = json.dumps({"chunks": [{"output_type": "text", "content": json.dumps(body)}]})
|
|
parsed = parse_skill_contract(raw)
|
|
assert parsed and parsed["skill_id"] == "sales-intelligence"
|
|
assert skill_minimum_for("receivables-cash-control") == ReasoningLevel.LOW
|
|
assert skill_minimum_for("root-cause-investigator") == ReasoningLevel.HIGH
|