From 8bd4b8e4130a57c17e0b3622ab300439d4139858 Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Thu, 20 Aug 2026 19:09:17 -0700 Subject: [PATCH] feat(session): record invocation provenance on the root session's mount plan A script firing `amplifier run --mode single "..."` produces a `session:start` record byte-identical (modulo ids and timestamps) to a human typing interactively. Downstream forensic tooling therefore has to guess from heuristics -- working-dir shape, prompt length, inter-prompt pacing -- and a large share of sessions land in an honest but useless UNKNOWN. This records what the CLI already knows at invocation time, at the one seam that already exists for it: `_inject_observability_events()` mutates the root mount plan immediately before `create_session()`, so a sibling `_inject_invocation_metadata()` lands beside it under the same ordering constraint. It rides the kernel's existing `session.metadata` passthrough channel, so there is no new event, no schema change, and no change needed in hooks-logging (`metadata` is not a promoted key, so it nests under `data`). mount_plan["session"]["metadata"]["invocation"] = { schema, mode, stdin_isatty, stdout_isatty, launched_by, launched_by_session_id, } Design notes worth keeping: - `mode` is the RESOLVED mode, not the raw `--mode` flag. The flag defaults to "single", so recording it would label an interactive session "single". Reaching interactive_chat() / execute_single() IS the resolution. - `stdout_isatty` is the cheap discriminator: a harness that fakes tty-ish pacing still usually redirects stdout. - `launched_by_session_id` covers only the CROSS-PROCESS launcher, from an AMPLIFIER_-prefixed env var. In-process lineage is already `parent_id`; duplicating it would create a second source of truth. Unset => null, never a fabricated id. - NO argv. Prompt text, `--api-key ...`, and `"$(cat token)"` all land in argv, and events.jsonl is long-lived and greppable. Raw config was already moved off session:start onto the redacted session:config event; a free-text argv field would reverse that. - Merged under the `invocation` key only -- caller/bundle metadata is untouched. - Self-report, not a security control. It defends against a harness that never thought about provenance, not one that lies. Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- amplifier_app_cli/main.py | 8 + amplifier_app_cli/session_runner.py | 101 +++++++++ tests/test_invocation_metadata.py | 334 ++++++++++++++++++++++++++++ 3 files changed, 443 insertions(+) create mode 100644 tests/test_invocation_metadata.py diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index 53e246fc..74391562 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -3480,6 +3480,10 @@ async def interactive_chat( bundle_name=bundle_name, initial_transcript=initial_transcript, prepared_bundle=prepared_bundle, + # Resolved mode: reaching interactive_chat() IS the resolution -- run.py + # dispatches here only after collapsing --mode, prompt presence, and pipe + # presence into "chat". + invocation_mode="chat", ) # Create fully initialized session (handles all setup including resume) @@ -4146,6 +4150,10 @@ async def execute_single( initial_transcript=initial_transcript, prepared_bundle=prepared_bundle, output_format=output_format, + # Resolved mode: reaching execute_single() IS the resolution -- run.py + # dispatches here for an explicit --mode single, and for a prompt or a + # pipe arriving with no --mode at all. + invocation_mode="single", ) # Create fully initialized session (handles all setup including resume) diff --git a/amplifier_app_cli/session_runner.py b/amplifier_app_cli/session_runner.py index 16cb6d2b..7eec70e0 100644 --- a/amplifier_app_cli/session_runner.py +++ b/amplifier_app_cli/session_runner.py @@ -26,6 +26,7 @@ from __future__ import annotations import logging +import os import sys import uuid from collections import Counter @@ -83,6 +84,13 @@ class SessionConfig: # Execution mode output_format: str = "text" # text | json | json-trace + # Resolved execution mode ("chat" | "single"), recorded as invocation + # provenance on session:start. This is the value AFTER prompt- and + # pipe-presence inference (commands/run.py), not the raw --mode flag: an + # interactive session launched without --mode would otherwise be recorded + # as "single". None => recorded as "unknown", never guessed. + invocation_mode: str | None = None + @property def is_resume(self) -> bool: """True if this is resuming an existing session.""" @@ -360,6 +368,93 @@ def _inject_observability_events(prepared_bundle: "PreparedBundle") -> None: inject_additional_events(prepared_bundle.mount_plan, _CLEANUP_EVENTS) +# Names the session that launched this OS process, when the launcher is a +# *different* process (an agent shelling out `amplifier run`, or foundation's +# subprocess spawn). In-process lineage is already carried end-to-end by +# ``parent_id``; this covers only the cross-process case ``parent_id`` cannot +# see. The ``AMPLIFIER_`` prefix is on foundation's subprocess env allowlist, +# so it propagates to children for free. Unset => null, never a fabricated id. +LAUNCHER_SESSION_ID_ENV = "AMPLIFIER_LAUNCHED_BY_SESSION_ID" + +# Bump only on a breaking change to the meaning of the fields below. +INVOCATION_SCHEMA = 1 + + +def _fd_isatty(fd: int) -> bool: + """``os.isatty`` that cannot take a session down. + + fd-level rather than ``sys.stdin.isatty()`` -- see dedicated_tty_input.py, + which already reasoned about this distinction: the fd is the thing that + actually matters, since sys.stdin can be swapped in-process. A closed or + invalid fd (daemonised harness) reads as "not a tty", which is the truth. + """ + try: + return os.isatty(fd) + except (OSError, ValueError): + return False + + +def _build_invocation_metadata(mode: str | None) -> dict[str, Any]: + """Describe HOW this session was invoked, for provenance consumers. + + Deliberately NOT recorded: argv / the full command line. Prompt text, + ``--api-key ...``, and ``"$(cat token)"`` all land in argv, while + events.jsonl is a long-lived, greppable, exported artifact. Raw config was + already moved *off* ``session:start`` onto the separate, redacted + ``session:config`` event -- adding an unredactable free-text field here + would reverse that. The fields below answer the provenance question + without opening that hole. + + This is a self-report, not a security control. A harness that fills these + in dishonestly will be believed, exactly as one that declines to pass + ``parent_id`` is believed. It defends against the overwhelmingly common + case -- a harness that never thought about provenance at all. + + Args: + mode: The *resolved* execution mode ("chat" / "single"), i.e. the value + after prompt- and pipe-presence inference, not the raw ``--mode`` + flag. ``None`` when a caller did not state one -- recorded as + "unknown" rather than guessed, since a wrong mode is worse than an + absent one. + """ + return { + "schema": INVOCATION_SCHEMA, + "mode": mode or "unknown", + "stdin_isatty": _fd_isatty(0), + "stdout_isatty": _fd_isatty(1), + "launched_by": "cli", + "launched_by_session_id": os.environ.get(LAUNCHER_SESSION_ID_ENV) or None, + } + + +def _inject_invocation_metadata( + prepared_bundle: "PreparedBundle", *, mode: str | None +) -> None: + """Record invocation provenance on the root session's mount plan. + + Rides the kernel's existing ``session.metadata`` passthrough channel + (amplifier-core ``docs/specs/CONTRIBUTION_CHANNELS.md``), so this surfaces + at ``session:start`` as ``metadata.invocation`` and lands in events.jsonl + at ``data.metadata.invocation``. No new event, no schema change, and no + change required in hooks-logging (``metadata`` is not a promoted key, so it + nests under ``data`` automatically). + + Merges under the ``invocation`` key only: any other metadata a bundle or + caller already placed on the mount plan is left exactly as it was. + + Args: + prepared_bundle: The PreparedBundle whose mount_plan will be updated + in-place. Same ordering constraint as + ``_inject_observability_events``: must run after + inject_user_providers() (step 4b) and before create_session() + (step 4c), which is when the kernel reads session.metadata. + mode: Resolved execution mode -- see ``_build_invocation_metadata``. + """ + session_section = prepared_bundle.mount_plan.setdefault("session", {}) + metadata = session_section.setdefault("metadata", {}) + metadata["invocation"] = _build_invocation_metadata(mode) + + async def _create_bundle_session( config: SessionConfig, session_id: str, @@ -400,6 +495,12 @@ async def _create_bundle_session( # config dict is populated when each hook module is mounted. _inject_observability_events(prepared_bundle) + # Step 4b-post: Record how this session was invoked, on the same mount plan + # and under the same ordering constraint. The kernel reads session.metadata + # when it builds the session:start payload, so this must land before + # create_session() below. + _inject_invocation_metadata(prepared_bundle, mode=config.invocation_mode) + # Step 4c: Create session (foundation handles init internally) # Self-healing: The kernel intentionally swallows module load errors to be resilient. # If providers fail to load due to stale install state (missing dependencies), diff --git a/tests/test_invocation_metadata.py b/tests/test_invocation_metadata.py new file mode 100644 index 00000000..08e8ab71 --- /dev/null +++ b/tests/test_invocation_metadata.py @@ -0,0 +1,334 @@ +"""Tests for invocation-provenance metadata on the root session's mount plan. + +app-cli records HOW a session was invoked (resolved mode, tty-ness of stdin and +stdout, launching component, cross-process launcher session id) under +``mount_plan["session"]["metadata"]["invocation"]``. The kernel's CP-SM +passthrough channel carries it to ``session:start``, where it persists as +``data.metadata.invocation`` in events.jsonl. + +What these tests pin down: + * the exact field set -- notably that argv is NOT among it + * the *resolved* mode, per entry point (chat vs single) + * both tty flags, mocked both ways + * pre-existing metadata is never clobbered + * an unset launcher env var yields null, never a fabricated id + * the mount plan is mutated BEFORE create_session() reads it +""" + +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from amplifier_app_cli.session_runner import ( + INVOCATION_SCHEMA, + LAUNCHER_SESSION_ID_ENV, + SessionConfig, + _build_invocation_metadata, + _inject_invocation_metadata, +) + +_MODULE = "amplifier_app_cli.session_runner" +_MAIN = "amplifier_app_cli.main" + +EXPECTED_FIELDS = { + "schema", + "mode", + "stdin_isatty", + "stdout_isatty", + "launched_by", + "launched_by_session_id", +} + + +def _prepared_bundle(mount_plan=None): + """A PreparedBundle stand-in carrying a real (mutable) mount plan dict.""" + bundle = MagicMock() + bundle.mount_plan = mount_plan if mount_plan is not None else {} + return bundle + + +def _invocation(bundle) -> dict: + return bundle.mount_plan["session"]["metadata"]["invocation"] + + +# --------------------------------------------------------------------------- +# Field set and shape +# --------------------------------------------------------------------------- + + +class TestInvocationShape: + def test_field_set_is_exactly_the_designed_one(self, monkeypatch): + monkeypatch.delenv(LAUNCHER_SESSION_ID_ENV, raising=False) + assert set(_build_invocation_metadata("single")) == EXPECTED_FIELDS + + def test_no_argv_or_command_line_is_recorded(self, monkeypatch): + """argv is a live secrets surface -- it must never reach events.jsonl. + + `amplifier run "$(cat prod-token.txt)"` and `--api-key ...` both land in + argv, and events.jsonl is long-lived, greppable, and exported. This + test exists so a future 'just add argv, it's useful' change has to + delete an explicit assertion rather than quietly widen the record. + """ + invocation = _build_invocation_metadata("single") + forbidden = ("argv", "cmd", "command", "args", "prompt", "env") + for key in invocation: + assert not any(bad in key.lower() for bad in forbidden), ( + f"Field {key!r} looks like a raw command-line/environment dump. " + f"Invocation provenance is deliberately limited to " + f"{sorted(EXPECTED_FIELDS)}." + ) + + def test_schema_is_versioned(self, monkeypatch): + monkeypatch.delenv(LAUNCHER_SESSION_ID_ENV, raising=False) + assert _build_invocation_metadata("single")["schema"] == INVOCATION_SCHEMA + assert isinstance(INVOCATION_SCHEMA, int) + + def test_launched_by_is_cli_from_this_construction_site(self): + assert _build_invocation_metadata("single")["launched_by"] == "cli" + + +# --------------------------------------------------------------------------- +# Resolved mode +# --------------------------------------------------------------------------- + + +class TestResolvedMode: + @pytest.mark.parametrize("mode", ["chat", "single"]) + def test_mode_recorded_verbatim(self, mode): + bundle = _prepared_bundle() + _inject_invocation_metadata(bundle, mode=mode) + assert _invocation(bundle)["mode"] == mode + + def test_absent_mode_is_unknown_not_guessed(self): + """A wrong mode is worse than an absent one (absent => UNKNOWN).""" + bundle = _prepared_bundle() + _inject_invocation_metadata(bundle, mode=None) + assert _invocation(bundle)["mode"] == "unknown" + + +# --------------------------------------------------------------------------- +# tty matrix +# --------------------------------------------------------------------------- + + +class TestTtyFlags: + @pytest.mark.parametrize( + ("stdin_tty", "stdout_tty"), + [(True, True), (True, False), (False, True), (False, False)], + ) + def test_isatty_matrix(self, stdin_tty, stdout_tty): + fds = {0: stdin_tty, 1: stdout_tty} + with patch(f"{_MODULE}.os.isatty", side_effect=lambda fd: fds[fd]): + invocation = _build_invocation_metadata("single") + assert invocation["stdin_isatty"] is stdin_tty + assert invocation["stdout_isatty"] is stdout_tty + + def test_closed_fd_reads_as_not_a_tty_instead_of_raising(self): + """A daemonised harness with a closed stdin must not fail to start.""" + with patch(f"{_MODULE}.os.isatty", side_effect=OSError("Bad file descriptor")): + invocation = _build_invocation_metadata("single") + assert invocation["stdin_isatty"] is False + assert invocation["stdout_isatty"] is False + + +# --------------------------------------------------------------------------- +# Cross-process launcher id +# --------------------------------------------------------------------------- + + +class TestLauncherSessionId: + def test_unset_env_var_yields_null_never_a_fabricated_id(self, monkeypatch): + monkeypatch.delenv(LAUNCHER_SESSION_ID_ENV, raising=False) + assert _build_invocation_metadata("single")["launched_by_session_id"] is None + + def test_empty_env_var_is_also_null(self, monkeypatch): + monkeypatch.setenv(LAUNCHER_SESSION_ID_ENV, "") + assert _build_invocation_metadata("single")["launched_by_session_id"] is None + + def test_set_env_var_is_recorded(self, monkeypatch): + monkeypatch.setenv(LAUNCHER_SESSION_ID_ENV, "launcher-session-abc") + assert ( + _build_invocation_metadata("single")["launched_by_session_id"] + == "launcher-session-abc" + ) + + def test_env_var_uses_the_amplifier_prefix(self): + """foundation's subprocess env allowlist forwards AMPLIFIER_* for free.""" + assert LAUNCHER_SESSION_ID_ENV.startswith("AMPLIFIER_") + + +# --------------------------------------------------------------------------- +# Merge behaviour +# --------------------------------------------------------------------------- + + +class TestMergeSemantics: + def test_existing_session_metadata_is_preserved(self): + bundle = _prepared_bundle( + {"session": {"metadata": {"agent_name": "kept", "run_id": "also-kept"}}} + ) + _inject_invocation_metadata(bundle, mode="single") + + metadata = bundle.mount_plan["session"]["metadata"] + assert metadata["agent_name"] == "kept" + assert metadata["run_id"] == "also-kept" + assert set(metadata) == {"agent_name", "run_id", "invocation"} + + def test_existing_session_section_keys_are_preserved(self): + bundle = _prepared_bundle( + {"session": {"orchestrator": "loop-basic", "context": "context-simple"}} + ) + _inject_invocation_metadata(bundle, mode="chat") + + session = bundle.mount_plan["session"] + assert session["orchestrator"] == "loop-basic" + assert session["context"] == "context-simple" + assert "invocation" in session["metadata"] + + def test_rest_of_mount_plan_is_untouched(self): + bundle = _prepared_bundle({"providers": [{"id": "anthropic"}], "tools": []}) + _inject_invocation_metadata(bundle, mode="single") + + assert bundle.mount_plan["providers"] == [{"id": "anthropic"}] + assert bundle.mount_plan["tools"] == [] + + def test_missing_session_section_is_created(self): + bundle = _prepared_bundle({}) + _inject_invocation_metadata(bundle, mode="single") + assert set(_invocation(bundle)) == EXPECTED_FIELDS + + +# --------------------------------------------------------------------------- +# Ordering -- the mount plan must be mutated before create_session() reads it +# --------------------------------------------------------------------------- + + +class TestOrdering: + @pytest.mark.anyio + async def test_mount_plan_carries_invocation_when_create_session_is_called( + self, tmp_path: Path + ): + """The kernel reads session.metadata during create_session(). + + Injecting after that call would be a silent no-op, so this asserts on + the mount plan as observed *at the moment* create_session() runs, not + afterwards. + """ + from amplifier_app_cli.session_runner import _create_bundle_session + + seen: dict = {} + + prepared_bundle = MagicMock() + prepared_bundle.mount_plan = {"providers": [], "tools": []} + + async def _capture_mount_plan(**_kwargs): + # Deep-ish copy of just what we assert on, captured at call time. + seen["invocation"] = dict( + prepared_bundle.mount_plan["session"]["metadata"]["invocation"] + ) + return MagicMock() + + prepared_bundle.create_session = AsyncMock(side_effect=_capture_mount_plan) + + cfg = SessionConfig( + config={}, + search_paths=[tmp_path], + verbose=False, + prepared_bundle=prepared_bundle, + bundle_name="test-bundle", + invocation_mode="single", + ) + + console = MagicMock() + console.status = MagicMock() + console.status.return_value.__enter__ = MagicMock() + console.status.return_value.__exit__ = MagicMock(return_value=False) + + with ( + patch(f"{_MODULE}.inject_user_providers", create=True), + patch(f"{_MODULE}._inject_observability_events"), + patch(f"{_MODULE}._should_attempt_self_healing", return_value=False), + patch("amplifier_app_cli.runtime.config.inject_user_providers"), + patch("amplifier_app_cli.lib.bundle_loader.AppModuleResolver"), + patch("amplifier_app_cli.paths.create_foundation_resolver"), + ): + await _create_bundle_session( + cfg, + session_id="test-session-id", + approval_system=MagicMock(), + display_system=MagicMock(), + console=console, + ) + + assert seen["invocation"]["mode"] == "single" + assert set(seen["invocation"]) == EXPECTED_FIELDS + + +# --------------------------------------------------------------------------- +# Entry points -- the two resolved-mode paths run.py dispatches to +# --------------------------------------------------------------------------- + + +class TestEntryPointsRecordResolvedMode: + """interactive_chat() => "chat"; execute_single() => "single". + + Reaching one of these functions IS the mode resolution: run.py collapses + --mode, prompt presence, and pipe presence before dispatching, so the raw + --mode flag (which defaults to "single") is never what gets recorded. + """ + + @pytest.mark.asyncio + async def test_execute_single_records_single(self, tmp_path: Path): + from amplifier_app_cli.main import execute_single + + captured: list[SessionConfig] = [] + + async def _capture(session_config, _console): + captured.append(session_config) + raise SystemExit(0) # stop before the rest of the session machinery + + with ( + patch( + f"{_MAIN}.create_initialized_session", + new=AsyncMock(side_effect=_capture), + ), + patch(f"{_MAIN}.console"), + pytest.raises(SystemExit), + ): + await execute_single( + prompt="Hi", + config={}, + search_paths=[tmp_path], + verbose=False, + bundle_name="test-bundle", + ) + + assert captured[0].invocation_mode == "single" + + @pytest.mark.asyncio + async def test_interactive_chat_records_chat(self, tmp_path: Path): + from amplifier_app_cli.main import interactive_chat + + captured: list[SessionConfig] = [] + + async def _capture(session_config, _console): + captured.append(session_config) + raise SystemExit(0) + + with ( + patch( + f"{_MAIN}.create_initialized_session", + new=AsyncMock(side_effect=_capture), + ), + patch(f"{_MAIN}.console"), + pytest.raises(SystemExit), + ): + await interactive_chat( + config={}, + search_paths=[tmp_path], + verbose=False, + bundle_name="test-bundle", + ) + + assert captured[0].invocation_mode == "chat"