Persist explicit internal session provenance across saves and resume

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"),

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)

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,

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

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,
     )
 
 

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.

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.