feat: add tag_session with Unicode sanitization (#670)

Stacked on #668 (`rename_session`).

## Changes

- NEW: `tag_session(session_id, tag, directory=None)` — appends a
`{type:'tag',tag:<tag>,sessionId:<id>}` JSONL entry. `list_sessions()`
reads the LAST tag from the file tail — most recent wins.
- Passing `None` appends an empty-string tag entry (`{"tag":""}`) which
readers treat as cleared.

## Port notes

**Unicode sanitization** (for CLI filter compat):
- Python's `re` module doesn't support `\p{Cf}` etc. without the
third-party `regex` module, so `_sanitize_unicode()` uses
`unicodedata.category(c) not in {"Cf","Co","Cn"}` per-character —
semantically equivalent to TS's `/[\p{Cf}\p{Co}\p{Cn}]/gu` regex (and
more reliable than TS's own ES-engine fallback).
- Applies NFKC normalization + explicit ranges (zero-width chars,
directional marks, BOM, private-use area) iteratively until stable (max
10 passes).

**Other behavior:**
- Tag is `.strip()`ed before storing; whitespace-only tags rejected
(`ValueError`).
- Tags that are pure invisible characters (e.g. only zero-width spaces)
are rejected after sanitization.
- Reuses `_append_to_session` from the `rename_session` PR (worktree
fallback, 0-byte stub skip, `O_APPEND` without `O_CREAT`).

<!-- CHANGELOG:START -->
- Add `tag_session()` for tagging sessions (with Unicode sanitization)
<!-- CHANGELOG:END -->

## Tests

17 new tests (TestTagSession, TestSanitizeUnicode). 298 total pass. Ruff
+ mypy clean.

Author: qing-ant

src/claude_agent_sdk/__init__.py

@@ -13,7 +13,7 @@
     CLINotFoundError,
     ProcessError,
 )
-from ._internal.session_mutations import rename_session
+from ._internal.session_mutations import rename_session, tag_session
 from ._internal.sessions import get_session_messages, list_sessions
 from ._internal.transport import Transport
 from ._version import __version__
@@ -423,6 +423,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> Any:
     "SessionMessage",
     # Session mutations
     "rename_session",
+    "tag_session",
     # Beta support
     "SdkBeta",
     # Sandbox support

src/claude_agent_sdk/_internal/session_mutations.py

@@ -22,6 +22,8 @@
 import errno
 import json
 import os
+import re
+import unicodedata
 from pathlib import Path
 
 from .sessions import (
@@ -92,6 +94,72 @@ def rename_session(
     _append_to_session(session_id, data, directory)
 
 
+def tag_session(
+    session_id: str,
+    tag: str | None,
+    directory: str | None = None,
+) -> None:
+    """Tag a session. Pass ``None`` to clear the tag.
+
+    Appends a ``{type:'tag',tag:<tag>,sessionId:<id>}`` JSONL entry.
+    ``list_sessions`` reads the LAST tag from the file tail — most recent
+    wins. Passing ``None`` appends an empty-string tag entry which
+    ``list_sessions`` treats as ``None`` (cleared).
+
+    Tags are Unicode-sanitized before storing (removes zero-width chars,
+    directional marks, private-use characters, etc.) for CLI filter
+    compatibility.
+
+    Args:
+        session_id: UUID of the session to tag.
+        tag: Tag string, or ``None`` to clear. Leading/trailing whitespace
+            is stripped. Must be non-empty after sanitization and stripping
+            (unless ``None``).
+        directory: Project directory path (same semantics as
+            ``list_sessions(directory=...)``). When omitted, all project
+            directories are searched for the session file.
+
+    Raises:
+        ValueError: If ``session_id`` is not a valid UUID, or if ``tag`` is
+            empty/whitespace-only after sanitization.
+        FileNotFoundError: If the session file cannot be found.
+
+    Example:
+        Tag a session::
+
+            tag_session(
+                "550e8400-e29b-41d4-a716-446655440000",
+                "experiment",
+                directory="/path/to/project",
+            )
+
+        Clear a tag::
+
+            tag_session(session_id, None)
+    """
+    if not _validate_uuid(session_id):
+        raise ValueError(f"Invalid session_id: {session_id}")
+    if tag is not None:
+        sanitized = _sanitize_unicode(tag).strip()
+        if not sanitized:
+            raise ValueError("tag must be non-empty (use None to clear)")
+        tag = sanitized
+
+    data = (
+        json.dumps(
+            {
+                "type": "tag",
+                "tag": tag if tag is not None else "",
+                "sessionId": session_id,
+            },
+            separators=(",", ":"),
+        )
+        + "\n"
+    )
+
+    _append_to_session(session_id, data, directory)
+
+
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
@@ -185,3 +253,49 @@ def _try_append(path: Path, data: str) -> bool:
         return True
     finally:
         os.close(fd)
+
+
+# ---------------------------------------------------------------------------
+# Unicode sanitization — ported from TS sanitization.ts
+# ---------------------------------------------------------------------------
+
+# Explicit ranges for dangerous Unicode characters. Python's regex supports
+# Unicode categories via \p{} only in the third-party `regex` module, so we
+# use explicit ranges here (matching the TS fallback paths).
+_UNICODE_STRIP_RE = re.compile(
+    "["
+    "\u200b-\u200f"  # Zero-width spaces, LTR/RTL marks
+    "\u202a-\u202e"  # Directional formatting characters
+    "\u2066-\u2069"  # Directional isolates
+    "\ufeff"  # Byte order mark
+    "\ue000-\uf8ff"  # Basic Multilingual Plane private use
+    "]"
+)
+
+# Format characters (Cf category) — the ones most commonly abused for
+# injection. We check this per-character since Python's re module doesn't
+# support \p{Cf} without the third-party regex module.
+_FORMAT_CATEGORIES = frozenset({"Cf", "Co", "Cn"})
+
+
+def _sanitize_unicode(value: str) -> str:
+    """Sanitize a string by removing dangerous Unicode characters.
+
+    Ported from TS ``partiallySanitizeUnicode``. Iteratively applies NFKC
+    normalization and strips format/private-use/unassigned characters until
+    no more changes occur (max 10 iterations).
+    """
+    current = value
+    for _ in range(10):
+        previous = current
+        # Apply NFKC normalization to handle composed character sequences
+        current = unicodedata.normalize("NFKC", current)
+        # Strip Cf (format), Co (private use), Cn (unassigned) categories
+        current = "".join(
+            c for c in current if unicodedata.category(c) not in _FORMAT_CATEGORIES
+        )
+        # Explicit ranges (redundant with category check but matches TS)
+        current = _UNICODE_STRIP_RE.sub("", current)
+        if current == previous:
+            break
+    return current

tests/test_session_mutations.py

@@ -1,4 +1,4 @@
-"""Tests for session mutation functions (rename_session, delete_session, tag_session)."""
+"""Tests for session mutation functions (rename_session, tag_session)."""
 
 from __future__ import annotations
 
@@ -9,8 +9,11 @@
 
 import pytest
 
-from claude_agent_sdk import list_sessions, rename_session
-from claude_agent_sdk._internal.session_mutations import _try_append
+from claude_agent_sdk import list_sessions, rename_session, tag_session
+from claude_agent_sdk._internal.session_mutations import (
+    _sanitize_unicode,
+    _try_append,
+)
 from claude_agent_sdk._internal.sessions import _sanitize_path
 
 # ---------------------------------------------------------------------------
@@ -253,3 +256,201 @@ def test_compact_json_format(self, claude_config_dir: Path, tmp_path: Path):
         assert lines[-1] == (
             f'{{"type":"custom-title","customTitle":"Title","sessionId":"{sid}"}}'
         )
+
+
+# ---------------------------------------------------------------------------
+# tag_session() tests
+# ---------------------------------------------------------------------------
+
+
+class TestTagSession:
+    """Tests for tag_session()."""
+
+    def test_invalid_session_id_raises(self, claude_config_dir: Path):
+        """Non-UUID session_id raises ValueError."""
+        with pytest.raises(ValueError, match="Invalid session_id"):
+            tag_session("not-a-uuid", "tag")
+        with pytest.raises(ValueError, match="Invalid session_id"):
+            tag_session("", "tag")
+
+    def test_empty_tag_raises(self, claude_config_dir: Path, tmp_path: Path):
+        """Empty or whitespace-only tag raises ValueError."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, _ = _make_session_file(project_dir)
+
+        with pytest.raises(ValueError, match="tag must be non-empty"):
+            tag_session(sid, "", directory=project_path)
+        with pytest.raises(ValueError, match="tag must be non-empty"):
+            tag_session(sid, "   ", directory=project_path)
+
+    def test_session_not_found_raises(self, claude_config_dir: Path, tmp_path: Path):
+        """Session not found raises FileNotFoundError."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        _make_project_dir(claude_config_dir, os.path.realpath(project_path))
+
+        sid = str(uuid.uuid4())
+        with pytest.raises(FileNotFoundError):
+            tag_session(sid, "tag", directory=project_path)
+
+    def test_appends_tag_entry(self, claude_config_dir: Path, tmp_path: Path):
+        """tag_session appends a {type:'tag'} JSON line."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, file_path = _make_session_file(project_dir)
+
+        tag_session(sid, "experiment", directory=project_path)
+
+        lines = file_path.read_text().strip().split("\n")
+        entry = json.loads(lines[-1])
+        assert entry["type"] == "tag"
+        assert entry["tag"] == "experiment"
+        assert entry["sessionId"] == sid
+
+    def test_tag_trimmed(self, claude_config_dir: Path, tmp_path: Path):
+        """Leading/trailing whitespace is stripped from tag."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, file_path = _make_session_file(project_dir)
+
+        tag_session(sid, "  my-tag  ", directory=project_path)
+
+        lines = file_path.read_text().strip().split("\n")
+        entry = json.loads(lines[-1])
+        assert entry["tag"] == "my-tag"
+
+    def test_none_clears_tag(self, claude_config_dir: Path, tmp_path: Path):
+        """Passing None appends an empty-string tag entry (clears tag)."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, file_path = _make_session_file(project_dir)
+
+        tag_session(sid, "original-tag", directory=project_path)
+        tag_session(sid, None, directory=project_path)
+
+        lines = file_path.read_text().strip().split("\n")
+        # Last entry is the clear
+        entry = json.loads(lines[-1])
+        assert entry["type"] == "tag"
+        assert entry["tag"] == ""
+        assert entry["sessionId"] == sid
+
+    def test_last_wins(self, claude_config_dir: Path, tmp_path: Path):
+        """Multiple tag calls — last one lands at EOF."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, file_path = _make_session_file(project_dir)
+
+        tag_session(sid, "first", directory=project_path)
+        tag_session(sid, "second", directory=project_path)
+        tag_session(sid, "third", directory=project_path)
+
+        lines = file_path.read_text().strip().split("\n")
+        entry = json.loads(lines[-1])
+        assert entry["tag"] == "third"
+        # All three tag entries present in file
+        tag_lines = [
+            json.loads(line) for line in lines if json.loads(line).get("type") == "tag"
+        ]
+        assert len(tag_lines) == 3
+
+    def test_compact_json_format(self, claude_config_dir: Path, tmp_path: Path):
+        """Appended JSON uses compact separators matching CLI."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, file_path = _make_session_file(project_dir)
+
+        tag_session(sid, "mytag", directory=project_path)
+
+        lines = file_path.read_text().strip().split("\n")
+        assert lines[-1] == f'{{"type":"tag","tag":"mytag","sessionId":"{sid}"}}'
+
+    def test_unicode_sanitization(self, claude_config_dir: Path, tmp_path: Path):
+        """Tag is sanitized: zero-width chars stripped."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, file_path = _make_session_file(project_dir)
+
+        # Tag with zero-width space and BOM embedded
+        dirty_tag = "clean\u200btag\ufeff"
+        tag_session(sid, dirty_tag, directory=project_path)
+
+        lines = file_path.read_text().strip().split("\n")
+        entry = json.loads(lines[-1])
+        assert entry["tag"] == "cleantag"
+
+    def test_sanitization_rejects_pure_invisible(
+        self, claude_config_dir: Path, tmp_path: Path
+    ):
+        """Tag that is only zero-width chars is rejected."""
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid, _ = _make_session_file(project_dir)
+
+        with pytest.raises(ValueError, match="tag must be non-empty"):
+            tag_session(sid, "\u200b\u200c\ufeff", directory=project_path)
+
+
+class TestSanitizeUnicode:
+    """Tests for the _sanitize_unicode helper."""
+
+    def test_passthrough_clean_string(self):
+        """Clean strings pass through unchanged."""
+        assert _sanitize_unicode("hello") == "hello"
+        assert _sanitize_unicode("tag-with-dashes_123") == "tag-with-dashes_123"
+
+    def test_strips_zero_width(self):
+        """Zero-width spaces/joiners are stripped."""
+        assert _sanitize_unicode("a\u200bb") == "ab"
+        assert _sanitize_unicode("a\u200cb") == "ab"  # zero-width non-joiner
+        assert _sanitize_unicode("a\u200db") == "ab"  # zero-width joiner
+
+    def test_strips_bom(self):
+        """Byte order mark is stripped."""
+        assert _sanitize_unicode("\ufeffhello") == "hello"
+
+    def test_strips_directional_marks(self):
+        """LTR/RTL marks and isolates are stripped."""
+        assert _sanitize_unicode("a\u202ab\u202cc") == "abc"
+        assert _sanitize_unicode("a\u2066b\u2069c") == "abc"
+
+    def test_strips_private_use(self):
+        """Private use area characters are stripped."""
+        assert _sanitize_unicode("a\ue000b") == "ab"
+        assert _sanitize_unicode("a\uf8ffb") == "ab"
+
+    def test_nfkc_normalization(self):
+        """NFKC normalization is applied (composed chars)."""
+        # Fullwidth 'A' → ASCII 'A'
+        assert _sanitize_unicode("\uff21") == "A"
+
+    def test_iterative_converges(self):
+        """Handles multi-pass cases safely (max 10 iterations)."""
+        # A string that needs multiple passes still converges
+        result = _sanitize_unicode("a" + "\u200b" * 20 + "b")
+        assert result == "ab"