Emit optional session catalog hints after native saves

AGENTS.md

@@ -182,3 +182,13 @@ The Highway watchdog records append-only `wake-needed` advisories only: it must
 not invoke Amplifier or consume markers, including triggers within its log
 verbosity gap. Keep its bounded fake-command tests proving each trigger records
 without starting host tmux or an Amplifier session.
+
+## Optional catalog discovery notices
+
+`catalog_hints.notify_session_saved()` runs only after a successful native save or
+metadata update. Keep it location-only, bounded, opt-in, and best effort; unavailable
+catalog state must never fail native persistence. Do not scan history, spawn a
+catalog, or write notices before the Foundation save succeeds. The catalog owns
+validation/indexing; native transcript, metadata, event and fork authority remain
+unchanged. Run `tests/test_catalog_hints.py` with session-store/metadata preservation
+tests when changing these seams.

README.md

@@ -444,6 +444,7 @@ toolkit/               # Standalone scenario tool utilities (at repo root)
 - [Agent Delegation](docs/AGENT_DELEGATION_IMPLEMENTATION.md) - Sub-session spawning and resumption
 - [Context Loading](docs/CONTEXT_LOADING.md) - @mention system implementation
 - [Interactive Mode](docs/INTERACTIVE_MODE.md) - REPL and slash commands
+- [Optional Session Catalog Discovery](docs/session-catalog-discovery.md) - Opt-in location-only notices after native saves, without a catalog dependency
 - [Architectural Decisions](docs/decisions/) - ADRs for major design choices
 
 **Authoritative Guides** (external, maintained in library repos):

amplifier_app_cli/catalog_hints.py

@@ -0,0 +1,56 @@
+"""Optional location-only notices for an independently running session catalog.
+
+The native save remains authoritative. No catalog dependency, process, network call,
+transcript read, or directory inventory belongs on this writer path.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import stat
+import uuid
+
+logger = logging.getLogger(__name__)
+HINT_DIRECTORY_ENV = "AMPLIFIER_SESSION_CATALOG_HINT_DIRECTORY"
+MAX_HINT_BYTES = 8192
+
+
+def notify_session_saved(session_directory: Path) -> bool:
+    """Coalesce a best-effort v1 notice after the caller's successful native save."""
+    configured = os.environ.get(HINT_DIRECTORY_ENV)
+    if not configured or os.name != "posix":
+        return False
+    temporary = None
+    try:
+        inbox = Path(configured).expanduser()
+        if not inbox.is_absolute() or inbox.is_symlink():
+            raise ValueError("Catalog hint directory must be absolute and not a symlink")
+        inbox.mkdir(parents=True, exist_ok=True, mode=0o700)
+        info = inbox.stat()
+        if not stat.S_ISDIR(info.st_mode) or info.st_mode & 0o077:
+            raise ValueError("Catalog hint directory must be private")
+        if hasattr(os, "getuid") and info.st_uid != os.getuid():
+            raise ValueError("Catalog hint directory has a different owner")
+        directory = str(session_directory.resolve(strict=True))
+        payload = json.dumps({"version": 1, "sessionDirectory": directory}, ensure_ascii=False).encode("utf-8")
+        if len(payload) > MAX_HINT_BYTES:
+            raise ValueError("Catalog hint exceeds its location-only bound")
+        key = hashlib.sha256(directory.encode("utf-8")).hexdigest()
+        temporary = inbox / ("." + key + "." + uuid.uuid4().hex + ".tmp")
+        descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(descriptor, "wb") as stream:
+            stream.write(payload)
+        os.replace(temporary, inbox / (key + ".json"))
+        return True
+    except (OSError, ValueError, RuntimeError) as exc:
+        logger.debug("Optional session catalog hint unavailable: %s", type(exc).__name__)
+        return False
+    finally:
+        if temporary is not None:
+            try:
+                temporary.unlink(missing_ok=True)
+            except OSError:
+                pass

amplifier_app_cli/session_store.py

@@ -23,6 +23,7 @@
 from amplifier_foundation.session.history import SessionHistoryStore
 from amplifier_foundation.session.metadata import SessionMetadataStore
 
+from amplifier_app_cli.catalog_hints import notify_session_saved
 from amplifier_app_cli.project_utils import get_project_slug
 from amplifier_foundation.paths.resolution import get_amplifier_home
 
@@ -129,6 +130,7 @@ def save(self, session_id: str, transcript: list, metadata: dict) -> None:
             transcript, redact_secrets(metadata), sanitizer=sanitize_message, merge_metadata=True
         )
 
+        notify_session_saved(session_dir)
         logger.debug(f"Session {session_id} saved successfully")
 
     def reserve_session(self, session_id: str) -> Path:
@@ -159,6 +161,7 @@ def save_new(self, session_id: str, transcript: list, metadata: dict) -> None:
             # write cannot delete another session's data.
             shutil.rmtree(session_dir, ignore_errors=True)
             raise
+        notify_session_saved(session_dir)
         logger.debug(f"New session {session_id} saved successfully")
 
     def _save_transcript(self, session_dir: Path, transcript: list) -> None:
@@ -171,10 +174,12 @@ def _save_transcript(self, session_dir: Path, transcript: list) -> None:
         SessionHistoryStore(session_dir).save_messages(
             transcript, sanitizer=sanitize_message
         )
+        notify_session_saved(session_dir)
 
     def _save_metadata(self, session_dir: Path, metadata: dict) -> None:
         """Save native metadata with the CLI's existing credential redaction."""
         SessionHistoryStore(session_dir).save_metadata(redact_secrets(metadata), merge_metadata=True)
+        notify_session_saved(session_dir)
 
     def load(self, session_id: str) -> tuple[list, dict]:
         """Load session state with corruption recovery.
@@ -254,14 +259,18 @@ def update_metadata(self, session_id: str, updates: dict) -> dict:
             raise FileNotFoundError(f"Session '{session_id}' not found")
 
         metadata = SessionMetadataStore(session_dir).update(redact_secrets(updates))
+        notify_session_saved(session_dir)
 
         logger.debug(f"Session {session_id} metadata updated: {list(updates.keys())}")
         return metadata
 
     def rename(self, session_id: str, name: str) -> dict:
         """Rename through Foundation without replacing transcript/runtime state."""
         self.get_metadata(session_id)  # Keep strict identity/existence validation.
-        return SessionMetadataStore(self.base_dir / session_id).set_name(name)
+        session_dir = self.base_dir / session_id
+        metadata = SessionMetadataStore(session_dir).set_name(name)
+        notify_session_saved(session_dir)
+        return metadata
 
     def get_metadata(self, session_id: str) -> dict:
         """Get session metadata without loading transcript.

docs/session-catalog-discovery.md

@@ -0,0 +1,38 @@
+# Optional session catalog discovery notices
+
+The CLI can notify an independently running `amplifier-session-catalog` after a
+successful native session save or metadata rename. This makes new/changed sessions
+visible sooner without a full metadata scan. It does not synchronize CLI editing
+state or change native history ownership.
+
+Configure a private absolute directory for both processes:
+
+```sh
+export AMPLIFIER_SESSION_CATALOG_HINT_DIRECTORY="$HOME/.amplifier/catalog-hints"
+amplifier-session-catalog serve --db /private/catalog.sqlite \
+  --home "$HOME/.amplifier" \
+  --hint-directory "$AMPLIFIER_SESSION_CATALOG_HINT_DIRECTORY" --scan-on-start
+```
+
+Run the CLI with the same environment setting. The configured catalog `--home` must
+match the CLI's native home; additional native homes can be supplied explicitly to
+the catalog. Its existing workspace/parent/internal-session visibility rules still
+apply. An ordinary root session becomes visible only when the catalog can establish
+its workspace, and child history stays behind its parent by default.
+
+Inbox v1 requires POSIX ownership/mode checks (macOS/Linux). On other platforms
+the optional writer remains disabled and ordinary catalog scanning still works.
+
+This integration is disabled by default and adds no CLI package dependency or
+background process. A notice contains only the absolute saved session directory and
+protocol version. It follows the catalog's public
+[writer hint inbox contract](https://github.com/microsoft/amplifier-session-catalog#optional-writer-hint-inbox-v1),
+using a bounded atomic notice coalesced by directory. It does not send transcript,
+metadata, event contents, secrets, or UI drafts. The catalog independently validates
+the location under its configured roots before reading bounded native metadata.
+
+A missing/unavailable inbox never turns a successful session save into an error.
+Notices are best effort; the catalog's periodic metadata scan repairs lost notices,
+and remains necessary for writers which do not emit them. The native transcript,
+metadata, event logs, backups, resume identity, and fork semantics stay authoritative
+and unchanged. No historical files need migration.

tests/test_catalog_hints.py

@@ -0,0 +1,76 @@
+"""Writer notices never own or change native history."""
+import hashlib
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+from amplifier_app_cli.catalog_hints import HINT_DIRECTORY_ENV, notify_session_saved
+from amplifier_app_cli.session_store import SessionStore
+
+pytestmark = pytest.mark.skipif(os.name != "posix", reason="Private hint inbox v1 requires POSIX ownership/mode checks")
+
+
+def test_opt_in_coalescing_and_saved_metadata(tmp_path, monkeypatch):
+    session = tmp_path / "session"
+    session.mkdir()
+    monkeypatch.delenv(HINT_DIRECTORY_ENV, raising=False)
+    assert notify_session_saved(session) is False
+    inbox = tmp_path / "hints"
+    monkeypatch.setenv(HINT_DIRECTORY_ENV, str(inbox))
+    for _ in range(25):
+        assert notify_session_saved(session)
+    notices = list(inbox.glob("*.json"))
+    assert len(notices) == 1
+    assert notices[0].name == hashlib.sha256(str(session.resolve()).encode()).hexdigest() + ".json"
+    assert json.loads(notices[0].read_text()) == {"version": 1, "sessionDirectory": str(session.resolve())}
+    assert len(list(inbox.iterdir())) == 1
+    assert notices[0].stat().st_mode & 0o077 == 0
+
+
+def test_save_rename_and_new_save_emit_only_after_native_success(tmp_path, monkeypatch):
+    inbox = tmp_path / "hints"
+    monkeypatch.setenv(HINT_DIRECTORY_ENV, str(inbox))
+    store = SessionStore(tmp_path / "sessions")
+    messages = [{"role": "user", "content": "preserved request"}]
+    store.save("one", messages, {"working_dir": str(tmp_path), "parent_id": None})
+    notice = next(inbox.glob("*.json"))
+    directory = Path(json.loads(notice.read_text())["sessionDirectory"])
+    transcript = (directory / "transcript.jsonl").read_bytes()
+    events = directory / "events.jsonl"
+    events.write_bytes(b'{"preserve":"all historical events"}\n')
+    notice.unlink()
+    store.rename("one", "New title")
+    assert notice.exists()
+    assert store.get_metadata("one")["name"] == "New title"
+    assert (directory / "transcript.jsonl").read_bytes() == transcript
+    assert events.read_bytes() == b'{"preserve":"all historical events"}\n'
+    store.save_new("two", messages, {"working_dir": str(tmp_path), "parent_id": "one"})
+    assert len(list(inbox.glob("*.json"))) == 2
+    with pytest.raises(FileExistsError):
+        store.save_new("two", messages, {})
+    assert len(list(inbox.glob("*.json"))) == 2
+    monkeypatch.setattr("amplifier_app_cli.session_store.SessionHistoryStore.save", lambda *a, **k: (_ for _ in ()).throw(OSError("native save failed")))
+    with pytest.raises(OSError, match="native save failed"):
+        store.save("failed", messages, {})
+    assert len(list(inbox.glob("*.json"))) == 2
+
+
+def test_unavailable_inbox_cannot_fail_native_save(tmp_path, monkeypatch):
+    blocked = tmp_path / "not-a-directory"
+    blocked.write_text("untouched")
+    monkeypatch.setenv(HINT_DIRECTORY_ENV, str(blocked))
+    store = SessionStore(tmp_path / "sessions")
+    store.save("safe", [{"role": "user", "content": "retained"}], {"working_dir": str(tmp_path)})
+    assert store.load("safe")[0][0]["content"] == "retained"
+    assert blocked.read_text() == "untouched"
+    monkeypatch.setenv(HINT_DIRECTORY_ENV, "relative-inbox")
+    assert notify_session_saved(tmp_path / "sessions/safe") is False
+    target = tmp_path / "actual"
+    target.mkdir(mode=0o700)
+    link = tmp_path / "link"
+    link.symlink_to(target, target_is_directory=True)
+    monkeypatch.setenv(HINT_DIRECTORY_ENV, str(link))
+    assert notify_session_saved(tmp_path / "sessions/safe") is False
+    assert list(target.iterdir()) == []