Merge pull request #71843 from NousResearch/bb/skill-title-leak

fix(sessions): stop a /skill's own text becoming the session title
This commit is contained in:
brooklyn! 2026-07-26 03:53:45 -05:00 committed by GitHub
commit d64ab9e553
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 685 additions and 32 deletions

View file

@ -54,6 +54,21 @@ _BUNDLE_MARKER = " skill bundle,"
_BUNDLE_USER_INSTRUCTION = "\nUser instruction: "
_BUNDLE_FIRST_SKILL_BLOCK = "\n\n[Loaded as part of the "
# The skill name sits in the first quoted span of the activation note, for both
# the single-skill and the bundle header ("work" / "/clean /work").
_SKILL_NAME_RE = re.compile(re.escape(_SKILL_INVOCATION_PREFIX) + r'"([^"]*)"')
# SQL LIKE pattern matching a skill-expanded turn, for listing queries that
# have to recognize scaffolding before the row reaches Python. The prefix
# contains no LIKE wildcards (`%`, `_`), so it needs no ESCAPE clause.
SKILL_SCAFFOLD_SQL_LIKE = _SKILL_INVOCATION_PREFIX + "%"
# Marks where a preview query joined the head and tail of a long scaffolded
# message. ``describe_skill_invocation`` may hand back a span that runs across
# the joint (a bundle instruction cut off by the head window); callers cut the
# description there rather than show the skill body on the far side.
SKILL_EXCERPT_JOINT = "\x1e"
def extract_user_instruction_from_skill_message(content: Any) -> Optional[str]:
"""Recover the user's instruction from a slash-skill-expanded turn.
@ -82,6 +97,41 @@ def extract_user_instruction_from_skill_message(content: Any) -> Optional[str]:
return None
def describe_skill_invocation(content: Any) -> Optional[str]:
"""Render a slash-skill-expanded turn the way the user typed it.
The expanded message embeds the whole skill body, so any surface that
summarizes a user turn from its raw content session titles, sidebar
previews, the ``/rewind`` picker otherwise shows the skill's own prose
as if the user had written it. That is how a skill's opening line ends up
as a session title.
Returns ``"/work — fix the title leak"``, or ``"/work"`` for a bare
invocation, or ``None`` when *content* is not skill scaffolding (the
caller should then summarize it as an ordinary message).
"""
if not isinstance(content, str) or not content.startswith(_SKILL_INVOCATION_PREFIX):
return None
match = _SKILL_NAME_RE.match(content)
name = (match.group(1) if match else "").strip()
# Bundle headers already carry their typed "/a /b" keys; a single skill is
# a bare name.
label = name if name.startswith("/") else f"/{name}"
instruction = extract_user_instruction_from_skill_message(content)
if instruction and instruction is not content:
# An excerpted message (head + tail, joined by SKILL_EXCERPT_JOINT) can
# put the joint inside the matched span — keep only the side the
# instruction marker was found on.
instruction = instruction.split(SKILL_EXCERPT_JOINT)[0]
instruction = " ".join(instruction.split())
if instruction:
return f"{label}{instruction}" if name else instruction
return label if name else None
def _extract_single_skill_user_instruction(message: str) -> Optional[str]:
# Single-skill format appends the user instruction after the skill body, so
# the last occurrence is the user-provided one; the body may quote this text.

View file

@ -71,6 +71,27 @@ def _auto_title_enabled() -> bool:
return True
def _summarize_user_message(user_message: str) -> str:
"""Collapse a slash-skill-expanded turn back to what the user typed.
A ``/skill`` invocation expands into a message that embeds the whole skill
body, so feeding it to the titler verbatim titles the session after the
*skill's* prose — "Kick off a task in a fresh isolated git worktree" — not
after the user's request. Reuse the canonical scaffolding parser so the
model sees ``/work fix the title leak`` instead.
"""
if not user_message:
return ""
try:
from agent.skill_commands import describe_skill_invocation
described = describe_skill_invocation(user_message)
except Exception:
logger.debug("Skill-scaffolding summary failed; titling raw", exc_info=True)
return user_message
return described if described is not None else user_message
def generate_title(
user_message: str,
assistant_response: str,
@ -110,7 +131,7 @@ def generate_title(
logger.debug("Title runtime validator raised; proceeding", exc_info=True)
# Truncate long messages to keep the request small
user_snippet = user_message[:500] if user_message else ""
user_snippet = _summarize_user_message(user_message)[:500]
assistant_snippet = assistant_response[:500] if assistant_response else ""
language = _title_language()
@ -143,6 +164,11 @@ def generate_title(
title = title.strip('"\'')
if title.lower().startswith("title:"):
title = title[6:].strip()
# A title is one line. A model that ignores "return ONLY the title" and
# answers the prompt instead (a shell transcript, a bulleted plan) would
# otherwise be stored verbatim and truncated mid-command. Keep the first
# non-empty line — the closest thing to a title in that response.
title = next((line.strip() for line in title.splitlines() if line.strip()), "")
# Enforce reasonable length
if len(title) > 80:
title = title[:77] + "..."

View file

@ -16072,6 +16072,29 @@ def main():
sessions_rename.add_argument("session_id", help="Session ID to rename")
sessions_rename.add_argument("title", nargs="+", help="New title for the session")
sessions_retitle = sessions_subparsers.add_parser(
"retitle-skills",
help="Re-title sessions whose auto-title came from a /skill's own text",
description=(
"Sessions opened with a /skill were auto-titled from the expanded "
"message, which embeds the whole skill body — so the title "
"describes the SKILL, not the request. This regenerates those "
"titles from what the user actually typed. Lists what it would "
"change unless --apply is passed."
),
)
sessions_retitle.add_argument(
"--apply",
action="store_true",
help="Write the new titles (default: dry run)",
)
sessions_retitle.add_argument(
"--limit",
type=int,
default=200,
help="Maximum sessions to examine (default: 200)",
)
sessions_browse = sessions_subparsers.add_parser(
"browse",
help="Interactive session picker — browse, search, and resume sessions",
@ -16952,6 +16975,67 @@ def main():
except ValueError as e:
print(f"Error: {e}")
elif action == "retitle-skills":
from agent.skill_commands import describe_skill_invocation
from agent.title_generator import generate_title
limit = max(1, int(getattr(args, "limit", 200) or 200))
apply_changes = bool(getattr(args, "apply", False))
def _is_titlelike(candidate: str) -> bool:
"""Reject a candidate that isn't a title at all.
An auxiliary model occasionally answers the prompt instead of
titling it and echoes the assistant's output ('$ df -h /'). The
live path has no alternative and takes what it gets, but this is
a REPAIR replacing a serviceable title with command output
would make things worse, so keep the old one.
"""
return bool(candidate) and candidate[0].isalnum()
candidates = db.list_skill_scaffolded_sessions(limit=limit)
if not candidates:
print("No sessions were titled from a /skill invocation.")
return
print(
f"{len(candidates)} session(s) opened with a /skill"
f"{'' if apply_changes else ' (dry run — pass --apply to write)'}:"
)
changed = 0
for row in candidates:
session_id = row["id"]
typed = describe_skill_invocation(row["content"]) or ""
first_reply = db.get_first_assistant_text(session_id) or ""
new_title = generate_title(typed, first_reply)
if not new_title or new_title == row["title"]:
continue
if not _is_titlelike(new_title):
print(f" {session_id}\n kept {row['title']!r} — got {new_title!r}")
continue
print(f" {session_id}\n {row['title']!r}\n{new_title!r}")
changed += 1
if not apply_changes:
continue
try:
db.set_session_title(session_id, new_title)
except ValueError:
# Unique-title collision. Dedupe the same way the live
# auto-titler does (base #2, base #3, ...) rather than
# leaving the leaked title in place.
deduped = db.get_next_title_in_lineage(new_title)
try:
db.set_session_title(session_id, deduped)
print(f" (renamed to {deduped!r} — title was taken)")
except ValueError as e:
print(f" skipped: {e}")
changed -= 1
if not changed:
print(" every title already reflects the user's request.")
elif apply_changes:
print(f"✓ Re-titled {changed} session(s).")
elif action == "browse":
limit = getattr(args, "limit", 500) or 500
source = getattr(args, "source", None)

View file

@ -29,6 +29,11 @@ from pathlib import Path
from agent.memory_manager import sanitize_context
from agent.message_sanitization import _sanitize_surrogates
from agent.skill_commands import (
SKILL_EXCERPT_JOINT,
SKILL_SCAFFOLD_SQL_LIKE,
describe_skill_invocation,
)
from hermes_constants import get_hermes_home
from hermes_cli.sqlite_runtime import (
is_sqlite_wal_reset_vulnerable as _is_sqlite_wal_reset_vulnerable,
@ -130,6 +135,51 @@ def _cwd_prefix_clause(cwd_prefix: str) -> Tuple[str, List[str]]:
return "(s.cwd = ? OR s.cwd LIKE ? OR s.cwd LIKE ?)", [prefix, f"{prefix}/%", f"{prefix}\\%"]
# Session preview = the head of the first user message, shown wherever a
# session has no title (sidebar rows, pickers, exports, the desktop's
# `sessionTitle` fallback).
#
# A /skill invocation expands into a message that embeds the whole skill body,
# so the plain head of it previews the SKILL's opening prose as if the user had
# written it. Scaffolded rows therefore carry a wider excerpt so
# ``_shape_preview`` can hand it to ``describe_skill_invocation`` and recover
# ``/work — fix the title leak``: the whole message while it stays under the
# budget, and head + tail (where the typed instruction lands) once it doesn't.
_PREVIEW_HEAD_CHARS = 63
_PREVIEW_SCAFFOLD_WINDOW = 400
_PREVIEW_MAX_CHARS = 60
_PREVIEW_CONTENT_SQL = "REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' ')"
_PREVIEW_SCAFFOLDED_SQL = f"m.content LIKE '{SKILL_SCAFFOLD_SQL_LIKE}'"
# The shared ``_preview_raw`` SELECT expression, interpolated by every listing
# query. A scaffolded row gets a wider excerpt: the whole message while it fits
# the budget, else head + tail (where the typed instruction lands) spliced
# around SKILL_EXCERPT_JOINT.
_PREVIEW_RAW_SELECT = (
f"CASE WHEN {_PREVIEW_SCAFFOLDED_SQL}"
f" AND LENGTH(m.content) > {_PREVIEW_SCAFFOLD_WINDOW * 2}"
f" THEN SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_SCAFFOLD_WINDOW})"
f" || '{SKILL_EXCERPT_JOINT}'"
f" || SUBSTR({_PREVIEW_CONTENT_SQL}, -{_PREVIEW_SCAFFOLD_WINDOW})"
f" WHEN {_PREVIEW_SCAFFOLDED_SQL}"
f" THEN SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_SCAFFOLD_WINDOW * 2})"
f" ELSE SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_HEAD_CHARS}) END"
)
def _shape_preview(raw: Any) -> str:
"""Turn a ``_preview_raw`` column into the short preview callers show."""
text = str(raw or "").strip()
if not text:
return ""
described = describe_skill_invocation(text)
text = described if described is not None else text.split(SKILL_EXCERPT_JOINT)[0]
if len(text) > _PREVIEW_MAX_CHARS:
return text[:_PREVIEW_MAX_CHARS] + "..."
return text
# A child session counts as a /branch (kept visible, never cascade-deleted) if
# it carries the stable marker OR the legacy end_reason heuristic holds.
_BRANCH_CHILD_SQL = (
@ -6005,7 +6055,7 @@ class SessionDB:
)
SELECT {_sel},
COALESCE(
(SELECT SUBSTR(REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' '), 1, 63)
(SELECT {_PREVIEW_RAW_SELECT}
FROM messages m
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
ORDER BY m.timestamp, m.id LIMIT 1),
@ -6030,7 +6080,7 @@ class SessionDB:
query = f"""
SELECT {_sel},
COALESCE(
(SELECT SUBSTR(REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' '), 1, 63)
(SELECT {_PREVIEW_RAW_SELECT}
FROM messages m
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
ORDER BY m.timestamp, m.id LIMIT 1),
@ -6052,13 +6102,7 @@ class SessionDB:
sessions = []
for row in rows:
s = dict(row)
# Build the preview from the raw substring
raw = s.pop("_preview_raw", "").strip()
if raw:
text = raw[:60]
s["preview"] = text + ("..." if len(raw) > 60 else "")
else:
s["preview"] = ""
s["preview"] = _shape_preview(s.pop("_preview_raw", ""))
# Drop the internal ordering column so callers see a clean dict.
s.pop("_effective_last_active", None)
sessions.append(s)
@ -6131,10 +6175,10 @@ class SessionDB:
# compute it generically rather than hardcoding the successor char.
prefix_hi = prefix[:-1] + chr(ord(prefix[-1]) + 1)
query = """
query = f"""
SELECT s.*,
COALESCE(
(SELECT SUBSTR(REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' '), 1, 63)
(SELECT {_PREVIEW_RAW_SELECT}
FROM messages m
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
ORDER BY m.timestamp, m.id LIMIT 1),
@ -6156,12 +6200,7 @@ class SessionDB:
runs: List[Dict[str, Any]] = []
for row in rows:
s = dict(row)
raw = s.pop("_preview_raw", "").strip()
if raw:
text = raw[:60]
s["preview"] = text + ("..." if len(raw) > 60 else "")
else:
s["preview"] = ""
s["preview"] = _shape_preview(s.pop("_preview_raw", ""))
runs.append(s)
return runs
@ -6177,7 +6216,7 @@ class SessionDB:
query = f"""
SELECT {_sel},
COALESCE(
(SELECT SUBSTR(REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' '), 1, 63)
(SELECT {_PREVIEW_RAW_SELECT}
FROM messages m
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
ORDER BY m.timestamp, m.id LIMIT 1),
@ -6196,14 +6235,54 @@ class SessionDB:
if not row:
return None
s = dict(row)
raw = s.pop("_preview_raw", "").strip()
if raw:
text = raw[:60]
s["preview"] = text + ("..." if len(raw) > 60 else "")
else:
s["preview"] = ""
s["preview"] = _shape_preview(s.pop("_preview_raw", ""))
return s
def list_skill_scaffolded_sessions(self, limit: int = 200) -> List[Dict[str, Any]]:
"""Titled sessions whose first user turn was a ``/skill`` invocation.
Those titles were generated from the expanded message, which embeds the
whole skill body so they describe the skill rather than the request.
Returns ``id``, ``title``, and the full first-turn ``content`` so a
caller can re-derive what the user typed. Newest first.
"""
with self._lock:
rows = self._conn.execute(
"""
SELECT s.id, s.title, m.content
FROM sessions s
JOIN messages m ON m.id = (
SELECT m2.id FROM messages m2
WHERE m2.session_id = s.id AND m2.role = 'user'
AND m2.content IS NOT NULL
ORDER BY m2.timestamp, m2.id LIMIT 1
)
WHERE s.title IS NOT NULL AND m.content LIKE ?
ORDER BY s.started_at DESC
LIMIT ?
""",
(SKILL_SCAFFOLD_SQL_LIKE, int(limit)),
).fetchall()
return [dict(row) for row in rows]
def get_first_assistant_text(self, session_id: str) -> str:
"""The session's first assistant reply as plain text ('' when none).
Pairs with :meth:`list_skill_scaffolded_sessions` so a re-title can feed
the titler the same (request, reply) shape the live path uses.
"""
with self._lock:
row = self._conn.execute(
"SELECT content FROM messages "
"WHERE session_id = ? AND role = 'assistant' AND content IS NOT NULL "
"ORDER BY timestamp, id LIMIT 1",
(session_id,),
).fetchone()
if not row:
return ""
decoded = self._decode_content(row["content"])
return decoded if isinstance(decoded, str) else ""
# =========================================================================
# Message storage
# =========================================================================
@ -7582,7 +7661,9 @@ class SessionDB:
if not preview:
preview = "[multimodal content]"
elif isinstance(decoded, str):
preview = decoded
# A /skill turn embeds the whole skill body; show what the user
# typed instead of the skill's opening prose.
preview = describe_skill_invocation(decoded) or decoded
else:
preview = ""
preview = " ".join(preview.split()) # collapse whitespace
@ -10311,10 +10392,10 @@ class SessionDB:
with self._lock:
try:
rows = self._conn.execute(
"""
f"""
SELECT s.*,
COALESCE(
(SELECT SUBSTR(REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' '), 1, 63)
(SELECT {_PREVIEW_RAW_SELECT}
FROM messages m
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
ORDER BY m.timestamp, m.id LIMIT 1),
@ -10340,10 +10421,10 @@ class SessionDB:
# telegram_dm_topic_bindings doesn't exist yet — no bindings
# means every telegram session for this user is "unlinked".
rows = self._conn.execute(
"""
f"""
SELECT s.*,
COALESCE(
(SELECT SUBSTR(REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' '), 1, 63)
(SELECT {_PREVIEW_RAW_SELECT}
FROM messages m
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
ORDER BY m.timestamp, m.id LIMIT 1),
@ -10365,8 +10446,7 @@ class SessionDB:
sessions: List[Dict[str, Any]] = []
for row in rows:
session = dict(row)
raw = str(session.pop("_preview_raw", "") or "").strip()
session["preview"] = raw[:60] + ("..." if len(raw) > 60 else "") if raw else ""
session["preview"] = _shape_preview(session.pop("_preview_raw", ""))
sessions.append(session)
return sessions

View file

@ -0,0 +1,164 @@
"""describe_skill_invocation() — recovering what the user typed from a /skill turn.
A /skill invocation expands into a message that embeds the whole skill body. Any
surface that summarizes a user turn from its raw content (session titles, sidebar
previews, the /rewind picker) otherwise shows the SKILL's prose as if the user had
written it.
Every case here builds the scaffolding with the real message builders rather than
a hand-written literal, so the description can't silently drift from the format it
parses.
"""
import pytest
import agent.skill_bundles as skill_bundles
import agent.skill_commands as skill_commands
import tools.skills_tool as skills_tool
from agent.skill_commands import (
SKILL_EXCERPT_JOINT,
SKILL_SCAFFOLD_SQL_LIKE,
describe_skill_invocation,
)
SKILL_BODY = "Kick off a task in a fresh isolated git worktree instead of the current checkout."
def _write_skill(skills_dir, name, body=SKILL_BODY):
skill_dir = skills_dir / name
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: Description for {name}\n---\n\n# {name}\n\n{body}\n"
)
return skill_dir
def _write_bundle(bundles_dir, slug, skills):
bundles_dir.mkdir(parents=True, exist_ok=True)
lines = [f"name: {slug}", "skills:"]
lines.extend(f" - {skill}" for skill in skills)
(bundles_dir / f"{slug}.yaml").write_text("\n".join(lines) + "\n")
@pytest.fixture()
def skills(tmp_path, monkeypatch):
"""Install a 'work' skill and a 'demo' bundle, with caches reset."""
skills_dir = tmp_path / "skills"
bundles_dir = tmp_path / "skill-bundles"
_write_skill(skills_dir, "work")
_write_skill(skills_dir, "clean", body="Polish your own diff by hand.")
_write_bundle(bundles_dir, "demo", ["work", "clean"])
monkeypatch.setattr(skills_tool, "SKILLS_DIR", skills_dir)
monkeypatch.setenv("HERMES_BUNDLES_DIR", str(bundles_dir))
monkeypatch.setattr(skill_commands, "_skill_commands", {})
monkeypatch.setattr(skill_commands, "_skill_commands_platform", None)
monkeypatch.setattr(skill_bundles, "_bundles_cache", {})
monkeypatch.setattr(skill_bundles, "_bundles_cache_mtime", None)
skill_commands.scan_skill_commands()
skill_bundles.scan_bundles()
return skills_dir
class TestDescribeSkillInvocation:
def test_passes_through_a_normal_message(self):
assert describe_skill_invocation("fix the title leak") is None
def test_ignores_non_string_content(self):
assert describe_skill_invocation(None) is None
assert describe_skill_invocation([{"type": "text", "text": "hi"}]) is None
def test_recovers_the_typed_instruction(self, skills):
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
assert describe_skill_invocation(message) == "/work — fix the title leak"
def test_bare_invocation_describes_the_skill_alone(self, skills):
message = skill_commands.build_skill_invocation_message("/work")
assert describe_skill_invocation(message) == "/work"
def test_never_surfaces_the_skill_body(self, skills):
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
described = describe_skill_invocation(message)
assert "worktree" not in described
assert "IMPORTANT" not in described
def test_runtime_note_is_not_part_of_the_instruction(self, skills):
message = skill_commands.build_skill_invocation_message(
"/work",
user_instruction="fix the title leak",
runtime_note="the runtime detail",
)
assert describe_skill_invocation(message) == "/work — fix the title leak"
def test_collapses_whitespace_in_a_multiline_instruction(self, skills):
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the\n title\tleak"
)
assert describe_skill_invocation(message) == "/work — fix the title leak"
def test_bundle_carries_its_typed_keys(self, skills):
result = skill_bundles.build_bundle_invocation_message(
"/demo", user_instruction="fix the title leak"
)
assert result is not None
message, _, _ = result
described = describe_skill_invocation(message)
assert described.endswith("— fix the title leak")
assert "worktree" not in described
def test_stacked_skills_describe_every_typed_key(self, skills):
result = skill_commands.build_stacked_skill_invocation_message(
["/work", "/clean"], user_instruction="fix the title leak"
)
assert result is not None
message, _, _ = result
assert describe_skill_invocation(message) == "/work /clean — fix the title leak"
class TestExcerptedScaffolding:
"""Preview queries hand over a head+tail excerpt, not the whole message."""
def _excerpt(self, message, window=400):
"""Mirror what _preview_raw_select() hands to _shape_preview()."""
flat = message.replace("\n", " ").replace("\r", " ")
if len(message) <= window * 2:
return flat[: window * 2]
return flat[:window] + SKILL_EXCERPT_JOINT + flat[-window:]
def test_excerpt_still_recovers_the_instruction(self, skills):
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
described = describe_skill_invocation(self._excerpt(message))
assert described == "/work — fix the title leak"
def test_description_never_runs_across_the_joint(self, skills):
# A long body pushes the head window into the middle of the skill text;
# the instruction is only present on the tail side.
skill_md = skills / "work" / "SKILL.md"
skill_md.write_text(skill_md.read_text().replace(SKILL_BODY, "filler line.\n" * 200))
skill_commands._skill_commands = {}
skill_commands._skill_commands_platform = None
skill_commands.scan_skill_commands()
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
described = describe_skill_invocation(self._excerpt(message))
assert SKILL_EXCERPT_JOINT not in described
assert "filler line" not in described
class TestSqlLikePattern:
def test_matches_the_prefix_the_builders_emit(self, skills):
message = skill_commands.build_skill_invocation_message("/work")
assert message.startswith(SKILL_SCAFFOLD_SQL_LIKE.rstrip("%"))
def test_carries_no_like_wildcards_needing_escape(self):
# The pattern is interpolated into SQL without an ESCAPE clause, so the
# literal part must not contain '%' or '_'.
assert "%" not in SKILL_SCAFFOLD_SQL_LIKE[:-1]
assert "_" not in SKILL_SCAFFOLD_SQL_LIKE[:-1]

View file

@ -212,6 +212,93 @@ class TestGenerateTitle:
user_content = captured_kwargs["messages"][1]["content"]
assert len(user_content) < 1100 # 500 + 500 + formatting
def test_skill_invocation_is_titled_from_what_the_user_typed(self):
"""A /skill turn embeds the whole skill body — the titler must never
see it, or the session gets named after the SKILL, not the request."""
skill_body = "Kick off a task in a fresh isolated git worktree. " * 20
expanded = (
'[IMPORTANT: The user has invoked the "work" skill, indicating they want '
"you to follow its instructions. The full skill content is loaded below.]\n\n"
f"{skill_body}\n\n"
"The user has provided the following instruction alongside the skill "
"invocation: fix the session title leak"
)
captured_kwargs = {}
def mock_call_llm(**kwargs):
captured_kwargs.update(kwargs)
resp = MagicMock()
resp.choices = [MagicMock()]
resp.choices[0].message.content = "Fixing The Session Title Leak"
return resp
with patch("agent.title_generator.call_llm", side_effect=mock_call_llm):
generate_title(expanded, "On it.")
sent = captured_kwargs["messages"][1]["content"]
assert "/work — fix the session title leak" in sent
assert "worktree" not in sent
assert "IMPORTANT" not in sent
def test_bare_skill_invocation_still_titles(self):
"""No typed instruction — the titler gets the command, not the body."""
expanded = (
'[IMPORTANT: The user has invoked the "weather-forecast-lookup" skill, '
"indicating they want you to follow its instructions. The full skill "
"content is loaded below.]\n\nPull clean multi-day forecasts."
)
captured_kwargs = {}
def mock_call_llm(**kwargs):
captured_kwargs.update(kwargs)
resp = MagicMock()
resp.choices = [MagicMock()]
resp.choices[0].message.content = "Weather Forecast Lookup"
return resp
with patch("agent.title_generator.call_llm", side_effect=mock_call_llm):
assert generate_title(expanded, "Here you go.") == "Weather Forecast Lookup"
sent = captured_kwargs["messages"][1]["content"]
assert "/weather-forecast-lookup" in sent
assert "Pull clean multi-day forecasts" not in sent
def test_plain_message_reaches_the_titler_unchanged(self):
"""The scaffolding summary must not touch an ordinary user turn."""
captured_kwargs = {}
def mock_call_llm(**kwargs):
captured_kwargs.update(kwargs)
resp = MagicMock()
resp.choices = [MagicMock()]
resp.choices[0].message.content = "A Title"
return resp
with patch("agent.title_generator.call_llm", side_effect=mock_call_llm):
generate_title("fix the session title leak", "On it.")
assert "fix the session title leak" in captured_kwargs["messages"][1]["content"]
def test_multiline_answer_collapses_to_its_first_line(self):
"""A model that answers the prompt instead of titling it must not have
a shell transcript stored as the session title."""
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = (
"macOS Disk Cleanup\n\n$ df -h /\nFilesystem Size Used Avail\n"
)
with patch("agent.title_generator.call_llm", return_value=mock_response):
assert generate_title("clean my disk", "Sure.") == "macOS Disk Cleanup"
def test_leading_blank_lines_do_not_empty_the_title(self):
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = "\n\n Real Title \nmore prose"
with patch("agent.title_generator.call_llm", return_value=mock_response):
assert generate_title("q", "a") == "Real Title"
def test_skips_when_title_generation_disabled(self):
"""auxiliary.title_generation.enabled=false disables automatic titles."""
config = {"auxiliary": {"title_generation": {"enabled": False}}}

View file

@ -0,0 +1,162 @@
"""Session previews must never surface a /skill's own body.
`preview` is the head of the first user message, and it is the TITLE FALLBACK on
every surface (sidebar rows, pickers, exports, desktop `sessionTitle`). A /skill
invocation expands into a message that embeds the whole skill body, so an
untitled skill session used to read `[IMPORTANT: The user has invoked the "work"
skill, indicatin...` in the sidebar.
These drive the real SQL/shaping path through SessionDB rather than calling the
shaper directly, so the CASE expression and the Python side are covered together.
"""
import pytest
import agent.skill_commands as skill_commands
import tools.skills_tool as skills_tool
from hermes_state import SessionDB
SKILL_BODY = (
"Kick off a task in a fresh isolated git worktree instead of the current checkout. "
"Look at what the repo already does, and copy it. Create the worktree and branch. "
)
@pytest.fixture()
def db(tmp_path):
session_db = SessionDB(db_path=tmp_path / "state.db")
yield session_db
session_db.close()
def _install_skill(tmp_path, monkeypatch, name="work", body=SKILL_BODY):
skills_dir = tmp_path / "skills"
skill_dir = skills_dir / name
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: Description for {name}\n---\n\n# {name}\n\n{body}\n"
)
monkeypatch.setattr(skills_tool, "SKILLS_DIR", skills_dir)
monkeypatch.setattr(skill_commands, "_skill_commands", {})
monkeypatch.setattr(skill_commands, "_skill_commands_platform", None)
skill_commands.scan_skill_commands()
return skills_dir
def _seed(db, session_id, content, *, title=None, reply="On it."):
db.create_session(session_id=session_id, source="cli", model="m")
db.append_message(session_id, role="user", content=content)
db.append_message(session_id, role="assistant", content=reply)
if title:
db.set_session_title(session_id, title)
class TestSkillPreview:
def test_plain_message_preview_is_unchanged(self, db):
_seed(db, "s1", "fix the title leak")
(row,) = db.list_sessions_rich(limit=10)
assert row["preview"] == "fix the title leak"
def test_long_plain_message_still_truncates(self, db):
_seed(db, "s1", "x" * 200)
(row,) = db.list_sessions_rich(limit=10)
assert row["preview"] == "x" * 60 + "..."
def test_skill_preview_shows_the_typed_instruction(self, db, tmp_path, monkeypatch):
_install_skill(tmp_path, monkeypatch)
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
_seed(db, "s1", message)
(row,) = db.list_sessions_rich(limit=10)
assert row["preview"] == "/work — fix the title leak"
def test_skill_preview_never_leaks_the_body(self, db, tmp_path, monkeypatch):
_install_skill(tmp_path, monkeypatch)
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
_seed(db, "s1", message)
(row,) = db.list_sessions_rich(limit=10)
assert "IMPORTANT" not in row["preview"]
assert "worktree" not in row["preview"]
def test_bare_skill_preview_is_the_command(self, db, tmp_path, monkeypatch):
_install_skill(tmp_path, monkeypatch)
message = skill_commands.build_skill_invocation_message("/work")
_seed(db, "s1", message)
(row,) = db.list_sessions_rich(limit=10)
assert row["preview"] == "/work"
def test_huge_skill_body_still_recovers_the_instruction(
self, db, tmp_path, monkeypatch
):
# Long enough that the head window lands mid-body and the SQL has to
# splice the tail to reach the instruction.
_install_skill(tmp_path, monkeypatch, body="filler line.\n" * 300)
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
_seed(db, "s1", message)
(row,) = db.list_sessions_rich(limit=10)
assert row["preview"] == "/work — fix the title leak"
assert "filler line" not in row["preview"]
def test_single_row_lookup_agrees_with_the_list(self, db, tmp_path, monkeypatch):
"""_get_session_rich_row shares the shaper — a compression tip must
surface the same preview the list does."""
_install_skill(tmp_path, monkeypatch)
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
_seed(db, "s1", message)
(listed,) = db.list_sessions_rich(limit=10)
single = db._get_session_rich_row("s1")
assert single["preview"] == listed["preview"] == "/work — fix the title leak"
def test_rewind_picker_shows_the_typed_instruction(
self, db, tmp_path, monkeypatch
):
_install_skill(tmp_path, monkeypatch)
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
_seed(db, "s1", message)
(entry,) = db.list_recent_user_messages("s1")
assert entry["preview"] == "/work — fix the title leak"
class TestSkillScaffoldedSessionLookup:
"""Backing queries for `hermes sessions retitle-skills`."""
def test_finds_only_titled_skill_sessions(self, db, tmp_path, monkeypatch):
_install_skill(tmp_path, monkeypatch)
message = skill_commands.build_skill_invocation_message(
"/work", user_instruction="fix the title leak"
)
_seed(db, "titled", message, title="Isolated Git Worktree Setup")
_seed(db, "untitled", message)
_seed(db, "plain", "fix the title leak", title="Fixing The Title Leak")
rows = db.list_skill_scaffolded_sessions()
assert [row["id"] for row in rows] == ["titled"]
assert rows[0]["title"] == "Isolated Git Worktree Setup"
# The full first turn comes back so the caller can re-derive the ask.
assert "fix the title leak" in rows[0]["content"]
def test_limit_is_honored(self, db, tmp_path, monkeypatch):
_install_skill(tmp_path, monkeypatch)
message = skill_commands.build_skill_invocation_message("/work")
for i in range(3):
_seed(db, f"s{i}", message, title=f"Title {i}")
assert len(db.list_skill_scaffolded_sessions(limit=2)) == 2
def test_first_assistant_text(self, db):
_seed(db, "s1", "hello", reply="first reply")
db.append_message("s1", role="assistant", content="second reply")
assert db.get_first_assistant_text("s1") == "first reply"
def test_first_assistant_text_missing_is_empty(self, db):
db.create_session(session_id="s1", source="cli", model="m")
db.append_message("s1", role="user", content="hello")
assert db.get_first_assistant_text("s1") == ""