This commit is contained in:
Samir Hanna Verza 2026-11-04 12:27:28 -03:00 • committed by GitHub
commit e3315e04e7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 108 additions and 3 deletions

View file

@ -297,7 +297,7 @@ def _update_note_impl(
try:
if note_id not in _notes_storage:
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
note = _notes_storage[note_id]
note = _notes_storage[note_id].copy()
if title is not None:
if not title.strip():
return {"success": False, "error": "Title cannot be empty"}
@ -312,6 +312,7 @@ def _update_note_impl(
except (ValueError, TypeError) as e:
return {"success": False, "error": f"Failed to update note: {e}"}
else:
_notes_storage[note_id].update(note)
_persist()
return {
"success": True,

View file

@ -238,7 +238,7 @@ def _apply_single_update(
) -> dict[str, Any] | None:
if todo_id not in agent_todos:
return {"todo_id": todo_id, "error": f"Todo with ID '{todo_id}' not found"}
todo = agent_todos[todo_id]
todo = agent_todos[todo_id].copy()
if title is not None:
if not title.strip():
return {"todo_id": todo_id, "error": "Title cannot be empty"}
@ -260,6 +260,7 @@ def _apply_single_update(
todo["status"] = status_candidate
todo["completed_at"] = datetime.now(UTC).isoformat() if status_candidate == "done" else None
todo["updated_at"] = datetime.now(UTC).isoformat()
agent_todos[todo_id].update(todo)
return None

View file

@ -2,16 +2,19 @@
from __future__ import annotations
import json
import uuid
from typing import TYPE_CHECKING
import pytest
from agents.tool_context import ToolContext
import strix.tools.notes.tools as notes_tools
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
@pytest.fixture(autouse=True)
@ -131,3 +134,51 @@ def test_list_notes_still_filters_on_real_values() -> None:
result = notes_tools._list_notes_impl(category="findings")
assert [n["title"] for n in result["notes"]] == ["recon"]
@pytest.mark.parametrize("content", ["", " "])
async def test_rejected_update_keeps_note_unchanged_after_later_save(
tmp_path: Path, content: str
) -> None:
notes_tools.hydrate_notes_from_disk(tmp_path)
note_id = notes_tools._create_note_impl("Original title", "Original evidence")["note_id"]
original = notes_tools._get_note_impl(note_id)["note"]
ctx = ToolContext(
context={"agent_id": "root"},
tool_name="update_note",
tool_call_id="call-update",
tool_arguments="{}",
)
raw = await notes_tools.update_note.on_invoke_tool(
ctx, json.dumps({"note_id": note_id, "title": "Rejected title", "content": content})
)
assert json.loads(raw)["success"] is False
assert notes_tools._get_note_impl(note_id)["note"] == original
# A later successful tool call persists every note, including this one.
notes_tools._create_note_impl("Another note", "Unrelated evidence")
notes_tools.hydrate_notes_from_disk(tmp_path)
assert notes_tools._get_note_impl(note_id)["note"] == original
def test_valid_note_update_persists_all_supplied_fields(tmp_path: Path) -> None:
notes_tools.hydrate_notes_from_disk(tmp_path)
note_id = notes_tools._create_note_impl(
"Original title", "Original evidence", tags=["old"], agent_id="root"
)["note_id"]
original = notes_tools._get_note_impl(note_id)["note"]
result = notes_tools._update_note_impl(
note_id, title=" Revised title ", content=" Revised evidence ", tags=[]
)
assert result["success"] is True
notes_tools.hydrate_notes_from_disk(tmp_path)
revised = notes_tools._get_note_impl(note_id)["note"]
assert revised["title"] == "Revised title"
assert revised["content"] == "Revised evidence"
assert revised["tags"] == []
assert revised["agent_id"] == "root"
assert revised["created_at"] == original["created_at"]
assert revised["updated_at"] >= original["updated_at"]

View file

@ -1,7 +1,7 @@
from __future__ import annotations
import json
from typing import Any
from typing import TYPE_CHECKING, Any
import pytest
from agents.tool_context import ToolContext
@ -10,6 +10,10 @@ from strix.tools.todo import tools
from strix.tools.todo.tools import _coerce_priority, create_todo
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolate_store() -> Any:
tools._todos_storage.clear()
@ -103,3 +107,51 @@ async def test_non_string_priority_does_not_fail_the_batch() -> None:
assert result["success"] is True
assert result["created_count"] == 3
assert {c["priority"] for c in result["created"]} == {"normal"}
@pytest.mark.parametrize(
"invalid_field", [{"priority": "invalid"}, {"priority": "high", "status": "invalid"}]
)
async def test_rejected_todo_update_is_not_persisted_with_valid_sibling(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, invalid_field: dict[str, str]
) -> None:
monkeypatch.setattr(tools, "_todos_path", None)
tools.hydrate_todos_from_disk(tmp_path)
created = await _create([{"title": "Original task"}, {"title": "Other task"}])
rejected_id, accepted_id = [item["todo_id"] for item in created["created"]]
original = tools._todos_storage["root"][rejected_id].copy()
ctx = ToolContext(
context={"agent_id": "root"},
tool_name="update_todo",
tool_call_id="call-update",
tool_arguments="{}",
)
raw = await tools.update_todo.on_invoke_tool(
ctx,
json.dumps(
{
"updates": json.dumps(
[
{
"todo_id": rejected_id,
"title": "Rejected title",
"description": "Rejected description",
**invalid_field,
},
{"todo_id": accepted_id, "status": "done"},
]
)
}
),
)
result = json.loads(raw)
assert result["success"] is False
assert result["updated"] == [accepted_id]
assert result["errors"][0]["todo_id"] == rejected_id
assert tools._todos_storage["root"][rejected_id] == original
tools.hydrate_todos_from_disk(tmp_path)
assert tools._todos_storage["root"][rejected_id] == original
assert tools._todos_storage["root"][accepted_id]["status"] == "done"
assert tools._todos_storage["root"][accepted_id]["completed_at"] is not None