Michael Chihlas
ac89e7b2fa
Server-assigns a uuid4 id to every AI-generated node (Finding 1 showstopper:
nodes had no id but the advance protocol keys on node_id, so ai_build walks
never advanced past question 1). Replaces the hidden {"node_type":"meta"}
walked_path convention with real category/problem_text/pending_node columns on
l1_walk_sessions (migration 61dda4f615c6) — fixes junk proposals + off-by-one
depth cap (Findings 8,9), and pending_node replays the served node on re-mount
(no duplicate paid LLM call). Intake honors explicit flow_id and adhoc=True
(Findings 4,5); flow_proposals.l1_session_id FK -> CASCADE (Finding 6 time
bomb); L1 category GET is owner+admin like PATCH and require_account_owner_or_admin
delegates to User.can_manage_account (Finding 7); escalate falls back to default
recipients + filters deleted_at + warns when empty (Finding 10). Cleanups: dead
ticket_ref removed, IntakeResponse per-outcome validator, unused acknowledged
dropped, escalations partial index, restored a deleted audit assertion.
Full Phase 2A backend set: 110 passed / 0 failed.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
105 lines
4.2 KiB
Python
105 lines
4.2 KiB
Python
import pytest
|
|
from app.services import ai_tree_builder as atb
|
|
|
|
|
|
class _FakeProvider:
|
|
def __init__(self, raw):
|
|
self._raw = raw
|
|
|
|
async def generate_json(self, *, system_prompt, messages, max_tokens):
|
|
return self._raw, None, None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_next_node_assigns_id_when_model_omits_it(monkeypatch):
|
|
"""The SYSTEM_PROMPT never asks the model for an id (Finding 1). The server
|
|
must assign one to every generated node, or the advance protocol — which keys
|
|
on node_id — can never record an answer and the walk stalls on question 1."""
|
|
monkeypatch.setattr(
|
|
atb, "get_ai_provider",
|
|
lambda *a, **k: _FakeProvider('{"node_type":"question","text":"Plugged in?"}'),
|
|
)
|
|
node = await atb.generate_next_node("printer down", "printer", [])
|
|
assert node["node_type"] == "question"
|
|
assert node.get("id"), "generated node must carry a server-assigned id"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_next_node_depth_cap_node_has_id(monkeypatch):
|
|
"""The depth-cap escalate node must also carry an id (it is persisted as
|
|
current_node_id and may be appended to walked_path)."""
|
|
walked = [{"node_type": "question", "id": f"n{i}", "text": "?", "answer": "no"}
|
|
for i in range(atb.MAX_DEPTH)]
|
|
node = await atb.generate_next_node("x", "printer", walked)
|
|
assert node["node_type"] == "escalate"
|
|
assert node.get("id")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_next_node_generation_failed_node_has_id(monkeypatch):
|
|
"""When both generation attempts fail, the fallback escalate node carries an id."""
|
|
monkeypatch.setattr(
|
|
atb, "get_ai_provider",
|
|
lambda *a, **k: _FakeProvider("not json at all"),
|
|
)
|
|
node = await atb.generate_next_node("x", "printer", [])
|
|
assert node["node_type"] == "escalate"
|
|
assert node["reason_category"] == "generation_failed"
|
|
assert node.get("id")
|
|
|
|
|
|
def test_validate_node_rejects_hard_floor_text():
|
|
node = {"node_type": "instruction", "id": "n1", "text": "Open regedit and change the key", "next": "generate"}
|
|
with pytest.raises(atb.UnsafeNodeError):
|
|
atb.validate_node(node)
|
|
|
|
|
|
def test_validate_node_accepts_safe_instruction():
|
|
node = {"node_type": "instruction", "id": "n1", "text": "Restart the printer.", "next": "generate"}
|
|
assert atb.validate_node(node)["node_type"] == "instruction"
|
|
|
|
|
|
def test_depth_cap_forces_escalate():
|
|
walked = [{"node_type": "question", "id": f"n{i}", "text": "?", "answer": "no"} for i in range(atb.MAX_DEPTH)]
|
|
node = atb.escalate_if_depth_exceeded(walked)
|
|
assert node is not None and node["node_type"] == "escalate"
|
|
|
|
|
|
def test_normalize_walked_path_builds_valid_tree():
|
|
walked = [
|
|
{"node_type": "question", "id": "n1", "text": "Powered on?", "answer": "no"},
|
|
{"node_type": "instruction", "id": "n2", "text": "Power it on.", "answer": "ack"},
|
|
{"node_type": "resolved", "id": "n3", "text": "Fixed."},
|
|
]
|
|
tree = atb.normalize_walked_path(walked)
|
|
assert isinstance(tree, dict) and tree.get("id") == "n1"
|
|
# untraversed 'yes' branch of n1 became a needs_review stub
|
|
assert any(n["node_type"] == "needs_review" for n in tree["nodes"].values())
|
|
|
|
|
|
def test_normalize_walk_ending_on_question_has_no_none_branches():
|
|
walked = [
|
|
{"node_type": "question", "id": "n1", "text": "Powered on?", "answer": "no"},
|
|
]
|
|
tree = atb.normalize_walked_path(walked)
|
|
n1 = tree["nodes"]["n1"]
|
|
assert n1["yes_next"] is not None and n1["no_next"] is not None
|
|
# both branches must reference real nodes present in the tree
|
|
assert n1["yes_next"] in tree["nodes"] and n1["no_next"] in tree["nodes"]
|
|
|
|
|
|
def test_normalize_preserves_escalate_reason_category():
|
|
walked = [
|
|
{"node_type": "question", "id": "n1", "text": "On?", "answer": "no"},
|
|
{"node_type": "escalate", "id": "n2", "text": "Beyond L1.",
|
|
"reason_category": "exhausted_safe_steps"},
|
|
]
|
|
tree = atb.normalize_walked_path(walked)
|
|
assert tree["nodes"]["n2"]["reason_category"] == "exhausted_safe_steps"
|
|
|
|
|
|
def test_normalize_empty_walk_returns_needs_review_root():
|
|
tree = atb.normalize_walked_path([])
|
|
assert tree["id"] in tree["nodes"]
|
|
assert tree["nodes"][tree["id"]]["node_type"] == "needs_review"
|