| from __future__ import annotations |
|
|
| import json |
| from pathlib import Path |
|
|
| from nexum_runtime.executor import ( |
| TOOL_CALL_END, |
| TOOL_CALL_START, |
| execute_tool_text, |
| tool_names, |
| tool_schemas, |
| ) |
| from nexum_runtime.tooling.drafting import DraftStore |
|
|
|
|
| def _call(body: str) -> str: |
| return f"{TOOL_CALL_START}{body}{TOOL_CALL_END}" |
|
|
|
|
| def test_drafting_tools_are_release_visible() -> None: |
| names = tool_names() |
| assert {"DraftCreate", "DraftCompare", "DraftSelect", "DraftStatus"} <= names |
| schema_names = { |
| str(row["function"]["name"]) |
| for row in tool_schemas() |
| if isinstance(row.get("function"), dict) |
| } |
| assert {"DraftCreate", "DraftCompare", "DraftSelect", "DraftStatus"} <= schema_names |
|
|
|
|
| def test_draft_compare_select_round_trip(tmp_path: Path) -> None: |
| session_id = "draft-session" |
| create = execute_tool_text( |
| _call( |
| "DraftCreate(" |
| "objective='repair current failure', " |
| "candidates=[{'name':'read-first','content':'Read the failing file'}, {'name':'run-first','content':'Run the failing command'}], " |
| "criteria=['uses observed evidence','avoids repeating failed action']" |
| ")" |
| ), |
| cwd=str(tmp_path), |
| session_id=session_id, |
| ) |
| assert len(create) == 1 |
| assert create[0].ok is True |
| created = json.loads(create[0].output) |
| draft_id = str(created["draft_id"]) |
| assert draft_id.startswith("draft_") |
| assert [row["index"] for row in created["candidates"]] == [0, 1] |
|
|
| compare = execute_tool_text( |
| _call( |
| "DraftCompare(" |
| f"draft_id='{draft_id}', " |
| "evidence_refs=['rep_0123456789abcdef0123456789abcdef'], " |
| "observations=['read-first preserves current evidence']" |
| ")" |
| ), |
| cwd=str(tmp_path), |
| session_id=session_id, |
| ) |
| assert compare[0].ok is True |
| compared = json.loads(compare[0].output) |
| assert len(compared["comparisons"]) == 1 |
|
|
| select = execute_tool_text( |
| _call( |
| "DraftSelect(" |
| f"draft_id='{draft_id}', " |
| "selected_index=0, " |
| "decision='best supported by current evidence', " |
| "evidence_refs=['rep_0123456789abcdef0123456789abcdef']" |
| ")" |
| ), |
| cwd=str(tmp_path), |
| session_id=session_id, |
| ) |
| assert select[0].ok is True |
| selected = json.loads(select[0].output) |
| assert selected["selection"]["selected_index"] == 0 |
|
|
| status = execute_tool_text( |
| _call(f"DraftStatus(draft_id='{draft_id}')"), |
| cwd=str(tmp_path), |
| session_id=session_id, |
| ) |
| assert status[0].ok is True |
| assert json.loads(status[0].output)["selection"]["decision"] == ( |
| "best supported by current evidence" |
| ) |
|
|
|
|
| def test_draft_receipts_are_session_scoped(tmp_path: Path) -> None: |
| record = DraftStore(tmp_path).create( |
| objective="choose action", |
| candidates=[{"name": "a", "content": "do a"}], |
| criteria=("evidence",), |
| session_id="session-a", |
| ) |
| assert DraftStore(tmp_path).get(record.draft_id, session_id="session-a").draft_id |
| result = execute_tool_text( |
| _call(f"DraftStatus(draft_id='{record.draft_id}')"), |
| cwd=str(tmp_path), |
| session_id="session-b", |
| ) |
| assert result[0].ok is False |
| assert "does not belong to this session" in result[0].error |
|
|
|
|
| def test_draft_sanitizes_workspace_path(tmp_path: Path) -> None: |
| workspace_file = tmp_path / "private.txt" |
| record = DraftStore(tmp_path).create( |
| objective=f"inspect {workspace_file}", |
| candidates=[{"name": "safe", "content": f"path {workspace_file}"}], |
| criteria=("evidence",), |
| session_id="safe-session", |
| ) |
| payload = json.dumps(record.to_dict(), sort_keys=True) |
| assert str(tmp_path) not in payload |
| assert record.objective == "inspect [WORKSPACE]/private.txt" |
|
|
