dinos
9285c6dad8
* feat(memory): migrate to profile memory files * chore(stream): read profile headings from templates * fix(display): keep assistant responses if response_text has started * fix(memory): do not treat failed bootstraps as profile creation * chore(memory): unlink blank legacy memory * fix(memory): resolve project_id once * fix(memory): preserve unreadable profile files * chore(tui): render streamed narration inline with tool timeline Update the TUI streaming timeline so assistant text emitted before or between tool calls is rendered inline where it occurs, rather than being kept as a single answer bubble above or below the tools. If the model begins an assistant response and then emits another tool call, the provisional response is converted into inline narration before that tool. The final assistant message then renders only the remaining response suffix, avoiding duplicate text in the completed transcript. Stop/cancel handling now preserves any active inline narration, appends the visible stopped marker only to the remaining displayed segment, and still returns the full normalized stopped response for channel callers. Completed tools continue to collapse while long runs are active, but expand again when the turn reaches a final state so the completed transcript shows the full tool timeline. * fix(stream): preserve narration around tool timelines Keep assistant narration attached to the tool call that follows it instead of folding all streamed text into the final answer block. Track narrated response segments in stream state, render them before their corresponding regular or task tool entries, and keep final answers limited to the response suffix that has not already been shown inline. Preserve narration across normal completion, stop/error final frames, sub-agent task calls, and collapsed live tool summaries. Add regression coverage for pending tools, completed tools, sub-agent task delegations, collapsed completed/running tool summaries, and final stop frames. * fix(tui): finalize inline narration transitions * test(memory): use canonical project id helper
331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""Tests for EvoScientist/stream/utils.py pure functions."""
|
|
|
|
from EvoScientist.stream.utils import (
|
|
_shorten_path,
|
|
count_lines,
|
|
format_tool_compact,
|
|
format_tool_compact_with_result,
|
|
has_args,
|
|
is_success,
|
|
truncate,
|
|
truncate_with_line_hint,
|
|
)
|
|
|
|
# === is_success ===
|
|
|
|
|
|
class TestIsSuccess:
|
|
def test_ok_prefix(self):
|
|
assert is_success("[OK] all good") is True
|
|
|
|
def test_failed_prefix(self):
|
|
assert is_success("[FAILED] bad") is False
|
|
|
|
def test_traceback(self):
|
|
assert is_success("Traceback (most recent call last)\n File ...") is False
|
|
|
|
def test_exception(self):
|
|
assert is_success("Exception: something went wrong") is False
|
|
|
|
def test_error(self):
|
|
assert is_success("Error: file not found") is False
|
|
|
|
def test_clean_output(self):
|
|
assert is_success("file1.py\nfile2.py") is True
|
|
|
|
def test_whitespace_stripped(self):
|
|
assert is_success(" [OK] with spaces ") is True
|
|
|
|
def test_error_in_code_content_not_false_positive(self):
|
|
# read_file returning code with "Error:" deep inside should be success
|
|
content = '#!/usr/bin/env python3\n"""\nSkill Packager\n"""\n\nprint(f"Error: not found")'
|
|
assert is_success(content) is True
|
|
|
|
def test_error_on_line4_not_false_positive(self):
|
|
content = "line1\nline2\nline3\nError: buried deep\nline5"
|
|
assert is_success(content) is True
|
|
|
|
def test_error_on_first_line(self):
|
|
assert is_success("Error: file not found\nsome detail") is False
|
|
|
|
def test_error_invoking_tool(self):
|
|
assert is_success("Error invoking tool 'write_file'") is False
|
|
|
|
def test_failed_to_uninstall(self):
|
|
assert (
|
|
is_success("Failed to uninstall skill: Skill not found: latex-paper-en")
|
|
is False
|
|
)
|
|
|
|
def test_failed_to_install(self):
|
|
assert is_success("Failed to install skill: git clone failed: ...") is False
|
|
|
|
def test_failed_in_code_content_not_false_positive(self):
|
|
content = '#!/usr/bin/env python3\n"""Helper"""\n\nif x:\n print("Failed to connect")'
|
|
assert is_success(content) is True
|
|
|
|
|
|
# === format_tool_compact ===
|
|
|
|
|
|
class TestFormatToolCompact:
|
|
def test_no_args(self):
|
|
assert format_tool_compact("execute", None) == "execute()"
|
|
assert format_tool_compact("execute", {}) == "execute()"
|
|
|
|
def test_execute(self):
|
|
result = format_tool_compact("execute", {"command": "ls -la"})
|
|
assert result == "execute(ls -la)"
|
|
|
|
def test_execute_long_command(self):
|
|
long_cmd = "x" * 60
|
|
result = format_tool_compact("execute", {"command": long_cmd})
|
|
assert len(result) < 70
|
|
assert result.endswith("\u2026)")
|
|
|
|
def test_read_file(self):
|
|
result = format_tool_compact("read_file", {"path": "src/main.py"})
|
|
assert result == "read_file(src/main.py)"
|
|
|
|
def test_write_file(self):
|
|
result = format_tool_compact("write_file", {"path": "out.txt"})
|
|
assert result == "write_file(out.txt)"
|
|
|
|
def test_edit_file(self):
|
|
result = format_tool_compact("edit_file", {"path": "f.py"})
|
|
assert result == "edit_file(f.py)"
|
|
|
|
# Global profile memory display (/memories/ = global)
|
|
def test_read_file_global_memory(self):
|
|
result = format_tool_compact(
|
|
"read_file", {"path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Reading memory"
|
|
|
|
def test_read_file_global_memory_file_path_alias(self):
|
|
result = format_tool_compact(
|
|
"read_file", {"file_path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Reading memory"
|
|
|
|
def test_read_file_any_global_memory_file(self):
|
|
result = format_tool_compact("read_file", {"path": "/memories/history.md"})
|
|
assert result == "Reading memory"
|
|
|
|
def test_write_file_global_memory(self):
|
|
result = format_tool_compact(
|
|
"write_file", {"path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Updating memory"
|
|
|
|
def test_edit_file_global_memory(self):
|
|
result = format_tool_compact(
|
|
"edit_file", {"path": "/memories/profile/USER_PROFILE.md"}
|
|
)
|
|
assert result == "Updating memory"
|
|
|
|
def test_write_edit_any_global_memory_file(self):
|
|
write_result = format_tool_compact("write_file", {"path": "/memories/soul.md"})
|
|
edit_result = format_tool_compact(
|
|
"edit_file", {"path": "/memories/skills-context.md"}
|
|
)
|
|
assert write_result == "Updating memory"
|
|
assert edit_result == "Updating memory"
|
|
|
|
# Project-local /memory/ files show normal tool display
|
|
def test_read_file_project_memory(self):
|
|
result = format_tool_compact(
|
|
"read_file", {"path": "/memory/ideation-memory.md"}
|
|
)
|
|
assert result == "read_file(/memory/ideation-memory.md)"
|
|
|
|
def test_edit_file_project_memory(self):
|
|
result = format_tool_compact(
|
|
"edit_file", {"path": "/memory/experiment-memory.md"}
|
|
)
|
|
assert result == "edit_file(/memory/experiment-memory.md)"
|
|
|
|
def test_memory_display_inferred_from_result_when_args_sparse(self):
|
|
read_result = format_tool_compact_with_result(
|
|
"read_file",
|
|
{},
|
|
"# User profile\n\nFounder: Zachary",
|
|
)
|
|
assert read_result == "Reading memory"
|
|
|
|
edit_result = format_tool_compact_with_result(
|
|
"edit_file",
|
|
{},
|
|
"Successfully replaced 1 instance(s) of the string in '/memories/profile/USER_PROFILE.md'",
|
|
)
|
|
assert edit_result == "Updating memory"
|
|
|
|
write_result = format_tool_compact_with_result(
|
|
"write_file",
|
|
{},
|
|
"Wrote updated content to '/memories/history.md'",
|
|
)
|
|
assert write_result == "Updating memory"
|
|
|
|
def test_profile_memory_inference_uses_profile_template_headings(self, monkeypatch):
|
|
from EvoScientist.middleware import memory
|
|
from EvoScientist.stream import utils
|
|
|
|
monkeypatch.setitem(
|
|
memory.PROFILE_TEMPLATES,
|
|
"/profile/CUSTOM.md",
|
|
"# Custom profile\n\n## Notes\n",
|
|
)
|
|
utils._profile_memory_headings.cache_clear()
|
|
|
|
try:
|
|
result = format_tool_compact_with_result(
|
|
"read_file",
|
|
{},
|
|
"# Custom profile\n\n- remembered",
|
|
)
|
|
finally:
|
|
utils._profile_memory_headings.cache_clear()
|
|
|
|
assert result == "Reading memory"
|
|
|
|
def test_project_memory_result_not_special(self):
|
|
result = format_tool_compact_with_result(
|
|
"write_file",
|
|
{},
|
|
"Wrote updated content to '/memory/ideation-memory.md'",
|
|
)
|
|
assert result != "Updating memory"
|
|
|
|
def test_glob(self):
|
|
result = format_tool_compact("glob", {"pattern": "*.py"})
|
|
assert result == "glob(*.py)"
|
|
|
|
def test_grep(self):
|
|
result = format_tool_compact("grep", {"pattern": "TODO", "path": "src/"})
|
|
assert result == "grep(TODO, src/)"
|
|
|
|
def test_ls(self):
|
|
assert format_tool_compact("ls", {"path": "/src"}) == "ls(/src)"
|
|
|
|
def test_write_todos_list(self):
|
|
todos = [{"status": "todo", "content": "a"}, {"status": "todo", "content": "b"}]
|
|
result = format_tool_compact("write_todos", {"todos": todos})
|
|
assert result == "write_todos(2 items)"
|
|
|
|
def test_write_todos_non_list(self):
|
|
result = format_tool_compact("write_todos", {"todos": "something"})
|
|
assert result == "write_todos(...)"
|
|
|
|
def test_read_todos(self):
|
|
assert format_tool_compact("read_todos", {}) == "read_todos()"
|
|
|
|
def test_task_with_type_and_desc(self):
|
|
result = format_tool_compact(
|
|
"task", {"subagent_type": "research-agent", "description": "Find papers"}
|
|
)
|
|
assert "Cooking with research-agent" in result
|
|
assert "Find papers" in result
|
|
|
|
def test_task_with_type_only(self):
|
|
result = format_tool_compact("task", {"subagent_type": "code-agent"})
|
|
assert result == "Cooking with code-agent"
|
|
|
|
def test_task_with_desc_only(self):
|
|
result = format_tool_compact("task", {"description": "do stuff"})
|
|
assert "Cooking with sub-agent" in result
|
|
|
|
def test_task_no_info(self):
|
|
result = format_tool_compact("task", {"other": "value"})
|
|
assert result == "Cooking with sub-agent"
|
|
|
|
def test_tavily_search(self):
|
|
result = format_tool_compact("tavily_search", {"query": "python testing"})
|
|
assert result == "tavily_search(python testing)"
|
|
|
|
def test_think_tool(self):
|
|
result = format_tool_compact("think_tool", {"reflection": "need more data"})
|
|
assert result == "think_tool(need more data)"
|
|
|
|
def test_unknown_tool(self):
|
|
result = format_tool_compact("custom_tool", {"key": "value"})
|
|
assert "custom_tool(" in result
|
|
assert "key=value" in result
|
|
|
|
def test_unknown_tool_long_value(self):
|
|
result = format_tool_compact("custom_tool", {"key": "a" * 30})
|
|
assert "\u2026" in result
|
|
|
|
|
|
# === truncate ===
|
|
|
|
|
|
class TestTruncate:
|
|
def test_within_limit(self):
|
|
assert truncate("hello", 10) == "hello"
|
|
|
|
def test_at_limit(self):
|
|
assert truncate("hello", 5) == "hello"
|
|
|
|
def test_over_limit(self):
|
|
result = truncate("hello world", 5)
|
|
assert result.startswith("hello")
|
|
assert "truncated" in result
|
|
|
|
|
|
# === _shorten_path ===
|
|
|
|
|
|
class TestShortenPath:
|
|
def test_short_path(self):
|
|
assert _shorten_path("src/main.py") == "src/main.py"
|
|
|
|
def test_long_path(self):
|
|
long_path = "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r.py"
|
|
result = _shorten_path(long_path, max_len=20)
|
|
assert result.startswith(".../")
|
|
assert result.endswith("r.py")
|
|
|
|
|
|
# === has_args ===
|
|
|
|
|
|
class TestHasArgs:
|
|
def test_none(self):
|
|
assert has_args(None) is False
|
|
|
|
def test_empty_dict(self):
|
|
assert has_args({}) is False
|
|
|
|
def test_non_empty(self):
|
|
assert has_args({"key": "val"}) is True
|
|
|
|
|
|
# === count_lines ===
|
|
|
|
|
|
class TestCountLines:
|
|
def test_empty(self):
|
|
assert count_lines("") == 0
|
|
|
|
def test_single_line(self):
|
|
assert count_lines("hello") == 1
|
|
|
|
def test_multi_line(self):
|
|
assert count_lines("a\nb\nc") == 3
|
|
|
|
|
|
# === truncate_with_line_hint ===
|
|
|
|
|
|
class TestTruncateWithLineHint:
|
|
def test_within_limit(self):
|
|
text, remaining = truncate_with_line_hint("a\nb\nc", max_lines=5)
|
|
assert remaining == 0
|
|
assert "a" in text
|
|
|
|
def test_over_limit(self):
|
|
text, remaining = truncate_with_line_hint("a\nb\nc\nd\ne\nf", max_lines=3)
|
|
assert remaining == 3
|
|
assert "d" not in text
|