diff --git a/amplifier_app_cli/commands/session.py b/amplifier_app_cli/commands/session.py index 1fdb9d97..dfe7b870 100644 --- a/amplifier_app_cli/commands/session.py +++ b/amplifier_app_cli/commands/session.py @@ -1043,6 +1043,7 @@ def sessions_fork( "forked_from_turn": result.forked_from_turn, "fork_cost_boundary": boundary, "forked_at": now, + "session_visibility": "chat", "created": now, "turn_count": count_turns(child_messages), "bundle": action_metadata.get("bundle"), diff --git a/amplifier_app_cli/incremental_save.py b/amplifier_app_cli/incremental_save.py index b1f6c11c..2d46b3ad 100644 --- a/amplifier_app_cli/incremental_save.py +++ b/amplifier_app_cli/incremental_save.py @@ -45,6 +45,7 @@ def __init__( bundle_name: str, config: dict[str, Any], root_state: Any = None, + creation_metadata: dict[str, str] | None = None, ): """Initialize incremental save hook. @@ -61,6 +62,7 @@ def __init__( self.bundle_name = bundle_name self.config = config self.root_state = root_state + self.creation_metadata = dict(creation_metadata or {}) self._last_message_count = 0 async def on_tool_post(self, event: str, data: dict[str, Any]): @@ -106,6 +108,7 @@ async def on_tool_post(self, event: str, data: dict[str, Any]): # Build metadata, preserving existing fields while updating dynamic ones metadata = { + **self.creation_metadata, **existing_metadata, # Preserve name, description, etc. "session_id": self.session_id, "created": existing_metadata.get( @@ -163,6 +166,7 @@ def register_incremental_save( bundle_name: str, config: dict[str, Any], root_state: Any = None, + creation_metadata: dict[str, str] | None = None, ) -> IncrementalSaveHook | None: """Register incremental save hook on session. @@ -185,7 +189,8 @@ def register_incremental_save( return None hook = IncrementalSaveHook( - session, store, session_id, bundle_name, config, root_state=root_state + session, store, session_id, bundle_name, config, root_state=root_state, + creation_metadata=creation_metadata, ) # Register with priority 900 (high, but below trace collector at 1000) diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index 042ef041..75557302 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -1998,6 +1998,7 @@ async def _fork_session(self, args: str) -> str: "forked_from_turn": result.forked_from_turn, "fork_cost_boundary": boundary, "forked_at": now, + "session_visibility": "chat", "created": now, "turn_count": count_turns(child_messages), "bundle": parent_metadata.get("bundle", self.bundle_name), @@ -2033,6 +2034,7 @@ async def _fork_session(self, args: str) -> str: "forked_from_turn": result.forked_from_turn, "fork_cost_boundary": boundary, "forked_at": now, + "session_visibility": "chat", "created": now, "turn_count": count_turns(child_messages), "bundle": parent_metadata.get("bundle", self.bundle_name), @@ -3889,6 +3891,7 @@ async def interactive_chat( bundle_name, config, root_state=vars(initialized).get("root_state"), + creation_metadata=vars(initialized).get("creation_metadata", {}), ) # Register /goal auto-continue progress renderer (docs/GOAL_COMMAND.md). @@ -3981,6 +3984,7 @@ async def _save_session(): # that may have been set by other hooks (e.g., session-naming) existing_metadata = store.get_metadata_if_exists(actual_session_id) metadata = { + **vars(initialized).get("creation_metadata", {}), **existing_metadata, # Preserve name, description, etc. "session_id": actual_session_id, "created": existing_metadata.get( @@ -4646,6 +4650,7 @@ async def _persist_session() -> int: # that may have been set by other hooks (e.g., session-naming) existing_metadata = store.get_metadata_if_exists(actual_session_id) metadata = { + **vars(initialized).get("creation_metadata", {}), **existing_metadata, # Preserve name, description, etc. "session_id": actual_session_id, "created": existing_metadata.get("created", datetime.now(UTC).isoformat()), @@ -4981,6 +4986,7 @@ async def save_for_handoff(): messages = await context.get_messages() store = SessionStore() metadata = { + **vars(initialized).get("creation_metadata", {}), **store.get_metadata_if_exists(actual_session_id), "session_id": actual_session_id, "bundle": bundle_name, diff --git a/amplifier_app_cli/session_provenance.py b/amplifier_app_cli/session_provenance.py new file mode 100644 index 00000000..53fbcebe --- /dev/null +++ b/amplifier_app_cli/session_provenance.py @@ -0,0 +1,35 @@ +"""Explicit creation provenance for native history presentation. + +These declarations are not permissions and are never inferred from invocation +mode, TTY state, prompts, titles, or the process that launched this CLI. +""" + +from __future__ import annotations + +import os +import re +from collections.abc import Mapping + +VISIBILITY_ENV = "AMPLIFIER_SESSION_VISIBILITY" +PURPOSE_ENV = "AMPLIFIER_SESSION_PURPOSE" + + +def creation_metadata( + *, is_resume: bool, env: Mapping[str, str] | None = None +) -> dict[str, str]: + """Capture a new root's declaration once; never relabel a resumed history. + + Unknown/missing declarations keep ordinary conversations visible. Purpose + is optional, bounded, and recorded only for explicitly internal jobs. + """ + if is_resume: + return {} + values = os.environ if env is None else env + visibility = values.get(VISIBILITY_ENV) + metadata = { + "session_visibility": "internal" if visibility == "internal" else "chat" + } + purpose = values.get(PURPOSE_ENV, "") + if visibility == "internal" and re.fullmatch(r"[a-z][a-z0-9_.-]{0,79}", purpose): + metadata["session_purpose"] = purpose + return metadata diff --git a/amplifier_app_cli/session_runner.py b/amplifier_app_cli/session_runner.py index 2b1aa1a7..8120bcbe 100644 --- a/amplifier_app_cli/session_runner.py +++ b/amplifier_app_cli/session_runner.py @@ -118,6 +118,7 @@ class InitializedSession: store: SessionStore = field(default_factory=SessionStore) configurator: Any = None root_state: Any = None + creation_metadata: dict[str, str] = field(default_factory=dict) async def cleanup(self, *, release_ownership: bool = True): """Clean up session resources.""" @@ -221,6 +222,16 @@ async def create_initialized_session( config.root_state = None raise + # Capture only at root creation, after the shared writer has detected any + # existing history. Resume must not inherit a launcher's new declaration. + from .session_provenance import creation_metadata + + provenance = ( + creation_metadata(is_resume=config.is_resume) + if inherited_root_id in (None, session_id) + else {} + ) + # Set root session metadata once — propagates to all child sessions via config deep-merge. # Guards ensure values are only stamped on first creation (root session); child sessions # inherit parent values via config deep-merge and the guards prevent overwriting them. @@ -253,6 +264,21 @@ async def create_initialized_session( # Step 4: Create session (bundle mode only) try: + # An internal job must be classified before session:start/CI discovery, + # including initialization failures. Exclusively create an empty native + # history; never replace an existing session or backfill legacy metadata. + if provenance.get("session_visibility") == "internal": + try: + SessionStore().save_new( + session_id, [], { + **provenance, + "session_id": session_id, + "bundle": config.bundle_name, + "working_dir": cwd, + }, + ) + except FileExistsError: + provenance = {} session = await _create_bundle_session( config=config, session_id=session_id, @@ -449,6 +475,7 @@ async def create_initialized_session( store=SessionStore(), configurator=configurator, root_state=config.root_state, + creation_metadata=provenance, ) diff --git a/docs/SESSION_PROVENANCE.md b/docs/SESSION_PROVENANCE.md new file mode 100644 index 00000000..9fee9dfd --- /dev/null +++ b/docs/SESSION_PROVENANCE.md @@ -0,0 +1,29 @@ +# Explicit internal session provenance + +A launcher that creates an implementation-only root job may set: + +```sh +AMPLIFIER_SESSION_VISIBILITY=internal \ +AMPLIFIER_SESSION_PURPOSE=memory.suggestion \ +amplifier run --output-format json '...' +``` + +The CLI records `session_visibility` and an optional bounded `session_purpose` in +native `metadata.json`. Purpose is a machine label matching +`[a-z][a-z0-9_.-]{0,79}`, not a prompt, credential, or permission. Missing or +unrecognized visibility means an ordinary `chat`. Agent origin, JSON mode, +launcher ancestry, and lack of a TTY do not imply an internal session. + +For a new internal root, an empty native history is created exclusively before +bundle initialization. This makes its classification available while the job +runs or if initialization fails. Existing history is never replaced by this +creation step. Incremental, final, failed-turn, and handoff saves preserve the +original fields. Resume does not read a new creation declaration from the +launcher's environment, including for unmarked legacy history. A deliberate +independent fork records `chat` without inheriting an internal parent's purpose. + +Consumers such as Amplifier Unified can hide explicitly internal histories from +ordinary chat lists while retaining diagnostic access. This is presentation +provenance, not authorization. Older consumers may ignore these fields; older +CLI versions do not persist the launcher's declaration. No legacy history is +reclassified or deleted. diff --git a/tests/test_headless_session_persistence.py b/tests/test_headless_session_persistence.py index e8b06489..a8c9b367 100644 --- a/tests/test_headless_session_persistence.py +++ b/tests/test_headless_session_persistence.py @@ -54,6 +54,7 @@ def get_capability(name: str): session.coordinator.cancellation.is_cancelled = False initialized = MagicMock() + initialized.creation_metadata = {"session_visibility": "internal", "session_purpose": "memory.suggestion"} initialized.session = session initialized.session_id = _SESSION_ID initialized.cleanup = AsyncMock() @@ -117,6 +118,8 @@ async def test_headless_json_output_creates_and_loads_new_session_from_amplifier transcript, metadata = SessionStore().load(_SESSION_ID) assert transcript[0]["content"] == "persist this" assert metadata["session_id"] == _SESSION_ID + assert metadata["session_visibility"] == "internal" + assert metadata["session_purpose"] == "memory.suggestion" assert SessionStore().base_dir.is_relative_to(isolated_home) assert not (default_home / ".amplifier").exists() @@ -291,6 +294,8 @@ async def cleanup(): transcript, metadata = SessionStore().load(_SESSION_ID) assert transcript == accumulated assert metadata["session_id"] == _SESSION_ID + assert metadata["session_visibility"] == "internal" + assert metadata["session_purpose"] == "memory.suggestion" assert metadata["turn_count"] == 1 # cleanup still ran exactly once after the save. diff --git a/tests/test_session_provenance.py b/tests/test_session_provenance.py new file mode 100644 index 00000000..698b689e --- /dev/null +++ b/tests/test_session_provenance.py @@ -0,0 +1,223 @@ +"""Explicit internal creation labels survive persistence without hiding human roots.""" + +from __future__ import annotations + +from contextlib import ExitStack +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from amplifier_app_cli.incremental_save import IncrementalSaveHook +from amplifier_app_cli.session_provenance import creation_metadata +from amplifier_app_cli.session_store import SessionStore + + +@pytest.mark.parametrize( + "env, expected", + [ + ({}, {"session_visibility": "chat"}), + ( + {"AMPLIFIER_SESSION_ORIGIN": "agent", "AMPLIFIER_EXECUTION_MODE": "single"}, + {"session_visibility": "chat"}, + ), + ({"AMPLIFIER_SESSION_VISIBILITY": "unknown"}, {"session_visibility": "chat"}), + ( + {"AMPLIFIER_SESSION_PURPOSE": "memory.suggestion"}, + {"session_visibility": "chat"}, + ), + ( + { + "AMPLIFIER_SESSION_VISIBILITY": "internal", + "AMPLIFIER_SESSION_PURPOSE": "memory.suggestion", + }, + {"session_visibility": "internal", "session_purpose": "memory.suggestion"}, + ), + ( + { + "AMPLIFIER_SESSION_VISIBILITY": "internal", + "AMPLIFIER_SESSION_PURPOSE": "private\ntext", + }, + {"session_visibility": "internal"}, + ), + ( + { + "AMPLIFIER_SESSION_VISIBILITY": "internal", + "AMPLIFIER_SESSION_PURPOSE": "a" * 81, + }, + {"session_visibility": "internal"}, + ), + ], +) +def test_only_explicit_internal_creation_is_hidden(env, expected): + assert creation_metadata(is_resume=False, env=env) == expected + assert creation_metadata(is_resume=True, env=env) == {} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "existing", + [ + None, + {}, + {"session_visibility": "chat"}, + {"session_visibility": "internal", "session_purpose": "original.job"}, + ], +) +async def test_incremental_save_preserves_creation_across_resume(tmp_path, existing): + store = SessionStore(base_dir=tmp_path / "sessions") + messages = [{"role": "user", "content": "synthetic"}] + if existing is not None: + store.save("root", messages, existing) + context = MagicMock(get_messages=AsyncMock(return_value=messages)) + session = MagicMock() + session.coordinator.get.return_value = context + new = creation_metadata( + is_resume=existing is not None, + env={ + "AMPLIFIER_SESSION_VISIBILITY": "internal", + "AMPLIFIER_SESSION_PURPOSE": "memory.suggestion", + }, + ) + hook = IncrementalSaveHook( + session, store, "root", "test", {}, creation_metadata=new + ) + await hook.on_tool_post("tool:post", {}) + transcript, metadata = store.load("root") + assert transcript == messages + assert { + key: metadata[key] + for key in ("session_visibility", "session_purpose") + if key in metadata + } == (existing if existing is not None else new) + + +@pytest.mark.anyio +@pytest.mark.parametrize("resume", [False, True]) +async def test_initializer_marks_before_start_without_relabelling_resume( + tmp_path, monkeypatch, resume +): + from amplifier_app_cli import session_runner + from tests.test_session_runner import ( + _configurator_patches, + _make_mock_session, + _make_session_config, + ) + + monkeypatch.setenv("AMPLIFIER_HOME", str(tmp_path / "home")) + monkeypatch.setenv("AMPLIFIER_SESSION_VISIBILITY", "internal") + monkeypatch.setenv("AMPLIFIER_SESSION_PURPOSE", "memory.suggestion") + native = SessionStore(base_dir=tmp_path / "sessions") + if resume: + native.save("root", [], {"session_visibility": "chat"}) + session = _make_mock_session() + config = _make_session_config( + session_id="root", initial_transcript=[] if resume else None + ) + + async def construct(**_kwargs): + metadata = native.get_metadata("root") + assert metadata["session_visibility"] == ("chat" if resume else "internal") + return session + + with ExitStack() as stack: + for patcher in _configurator_patches(session): + stack.enter_context(patcher) + stack.enter_context( + patch.object(session_runner, "SessionStore", return_value=native) + ) + stack.enter_context( + patch.object( + session_runner, "_create_bundle_session", side_effect=construct + ) + ) + initialized = await session_runner.create_initialized_session( + config, MagicMock() + ) + assert initialized.creation_metadata == ( + {} + if resume + else {"session_visibility": "internal", "session_purpose": "memory.suggestion"} + ) + + +@pytest.mark.anyio +async def test_initialization_failure_keeps_internal_marker_and_releases_owner( + tmp_path, monkeypatch +): + from amplifier_app_cli import session_runner + from tests.test_session_runner import ( + _configurator_patches, + _make_mock_session, + _make_session_config, + ) + + monkeypatch.setenv("AMPLIFIER_HOME", str(tmp_path / "home")) + monkeypatch.setenv("AMPLIFIER_SESSION_VISIBILITY", "internal") + native = SessionStore(base_dir=tmp_path / "sessions") + root = MagicMock() + config = _make_session_config(session_id="root", root_state=root) + with ExitStack() as stack: + for patcher in _configurator_patches(_make_mock_session()): + stack.enter_context(patcher) + stack.enter_context( + patch.object(session_runner, "SessionStore", return_value=native) + ) + stack.enter_context( + patch.object( + session_runner, + "_create_bundle_session", + side_effect=RuntimeError("synthetic init failure"), + ) + ) + with pytest.raises(RuntimeError, match="synthetic init failure"): + await session_runner.create_initialized_session(config, MagicMock()) + assert native.load("root")[1]["session_visibility"] == "internal" + root.release.assert_called_once() + + +@pytest.mark.anyio +@pytest.mark.parametrize("kind", ["legacy", "chat", "child"]) +async def test_existing_or_child_identity_is_not_relabelled_before_creation( + tmp_path, monkeypatch, kind +): + from amplifier_app_cli import session_runner + from tests.test_session_runner import ( + _configurator_patches, + _make_mock_session, + _make_session_config, + ) + + monkeypatch.setenv("AMPLIFIER_HOME", str(tmp_path / "home")) + monkeypatch.setenv("AMPLIFIER_SESSION_VISIBILITY", "internal") + native = SessionStore(base_dir=tmp_path / "sessions") + if kind != "child": + native.save( + "root", + [{"role": "user", "content": "original"}], + {"session_visibility": "chat"} if kind == "chat" else {}, + ) + before = { + path: path.read_bytes() for path in native.base_dir.rglob("*") if path.is_file() + } + config = _make_session_config( + session_id="root", + session_config_initial={"root_session_id": "parent"} if kind == "child" else {}, + ) + with ExitStack() as stack: + for patcher in _configurator_patches(_make_mock_session()): + stack.enter_context(patcher) + stack.enter_context( + patch.object(session_runner, "SessionStore", return_value=native) + ) + stack.enter_context( + patch.object( + session_runner, + "_create_bundle_session", + side_effect=RuntimeError("stop before execution"), + ) + ) + with pytest.raises(RuntimeError, match="stop before execution"): + await session_runner.create_initialized_session(config, MagicMock()) + assert before == { + path: path.read_bytes() for path in native.base_dir.rglob("*") if path.is_file() + } diff --git a/tests/test_shared_root_lifecycle.py b/tests/test_shared_root_lifecycle.py index d26c68b1..31662064 100644 --- a/tests/test_shared_root_lifecycle.py +++ b/tests/test_shared_root_lifecycle.py @@ -76,7 +76,8 @@ def test_session_fork_reads_latest_native_history_over_legacy_checkpoint( native.save( "shared-root", _messages("stale projection"), - {"session_id": "shared-root", "bundle": "bundle:stale"}, + {"session_id": "shared-root", "bundle": "bundle:stale", + "session_visibility": "internal", "session_purpose": "memory.suggestion"}, ) parent_capture = native.base_dir / "shared-root" / "context-intelligence" / "events.jsonl" parent_capture.parent.mkdir() @@ -97,6 +98,8 @@ def test_session_fork_reads_latest_native_history_over_legacy_checkpoint( assert metadata["bundle"] == "bundle:stale" assert metadata["parent_id"] == "shared-root" assert metadata["forked_from_turn"] == 1 + assert metadata["session_visibility"] == "chat" + assert "session_purpose" not in metadata assert metadata["fork_cost_boundary"]["status"] == "verified" assert metadata["fork_cost_boundary"]["cumulative_cost_usd_by_turn"] == ["0.10"] @@ -286,7 +289,8 @@ async def test_live_fork_uses_current_context_without_reacquiring_shared_lock( root_handle = MagicMock() root_handle.read.return_value = ( _messages("held metadata authority"), - {"bundle": "bundle:held", "model": "held-model"}, + {"bundle": "bundle:held", "model": "held-model", + "session_visibility": "internal", "session_purpose": "memory.suggestion"}, ) def get_capability(name: str): @@ -319,6 +323,8 @@ def get_capability(name: str): assert transcript[0]["content"] == "live authority" assert metadata["parent_id"] == "shared-root" assert metadata["bundle"] == "bundle:held" + assert metadata["session_visibility"] == "chat" + assert "session_purpose" not in metadata assert metadata["fork_cost_boundary"]["status"] == "unavailable" root_handle.read.assert_called_once_with(native)