From 9e653906f8dcf62d39c8272d6a890a637d28ab63 Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:35:49 -0700 Subject: [PATCH 1/2] feat: add passive CLI doctor diagnostics Add a bounded provenance collector with a thin Click command, an on-demand diagnostic agent, compatibility-preserving entry points, and installed-wheel safety coverage. Generated with Amplifier Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .github/workflows/ci.yml | 11 + AGENTS.md | 17 + README.md | 42 ++ agents/cli-doctor.md | 102 +++++ amplifier_app_cli/__init__.py | 52 ++- amplifier_app_cli/__main__.py | 18 +- amplifier_app_cli/commands/__init__.py | 67 +++- amplifier_app_cli/commands/doctor.py | 59 +++ amplifier_app_cli/lib/diagnostics.py | 515 +++++++++++++++++++++++++ amplifier_app_cli/main.py | 381 ++++++++++-------- behaviors/cli-expertise.yaml | 1 + pyproject.toml | 2 +- tests/check_doctor_wheel.py | 211 ++++++++++ tests/doctor_safety_probe.py | 125 ++++++ tests/test_amplifier_home_isolation.py | 16 + tests/test_cli_doctor_agent.py | 62 +++ tests/test_doctor.py | 359 +++++++++++++++++ tests/test_doctor_entrypoint.py | 514 ++++++++++++++++++++++++ tests/test_shell_completion.py | 7 + 19 files changed, 2379 insertions(+), 182 deletions(-) create mode 100644 agents/cli-doctor.md create mode 100644 amplifier_app_cli/commands/doctor.py create mode 100644 amplifier_app_cli/lib/diagnostics.py create mode 100644 tests/check_doctor_wheel.py create mode 100644 tests/doctor_safety_probe.py create mode 100644 tests/test_cli_doctor_agent.py create mode 100644 tests/test_doctor.py create mode 100644 tests/test_doctor_entrypoint.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 07daf4cf..4688eb59 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,6 +39,17 @@ jobs: # watching, which is how a total macOS breakage (#247) shipped. run: uv run pytest -q + - name: Verify installed doctor wheel and generated legacy launcher outside checkout + if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.11' + shell: bash + run: | + uv build --wheel --out-dir "$RUNNER_TEMP/doctor-wheels" + uv venv "$RUNNER_TEMP/doctor-runtime" + uv pip install --python "$RUNNER_TEMP/doctor-runtime/bin/python" "$RUNNER_TEMP"/doctor-wheels/*.whl + cp tests/check_doctor_wheel.py tests/doctor_safety_probe.py "$RUNNER_TEMP/" + cd "$RUNNER_TEMP" + env -u PYTHONPATH -u PYTHONDONTWRITEBYTECODE "$RUNNER_TEMP/doctor-runtime/bin/python" check_doctor_wheel.py + integration: name: pytest -m integration (${{ matrix.os }}) runs-on: ${{ matrix.os }} diff --git a/AGENTS.md b/AGENTS.md index 0002f34d..cff6eb06 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -257,3 +257,20 @@ 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. + +## Passive doctor entry point + +The published console script, legacy `main:main` launcher and module entry +dispatch exact leading `doctor` before normal CLI startup. Keep collection free +of key loading, preparation, +network, subprocesses and application-state writes; Python's initial import +cache is a separate interpreter boundary. Preserve ownership enforcement and +ordinary startup/TLS behavior. Per-scope declarations are not effective settings +or live-session evidence. Keep public package/command exports static across +argv routes; deferred `main.py` imports must initialize the real Click group +once on explicit `cli` access or later ordinary dispatch, in the same module +namespace so existing global patches still apply. Unknown attribute probes must +not initialize it. Run `tests/test_doctor.py`, +`tests/test_doctor_entrypoint.py`, `tests/test_cli_doctor_agent.py` and the +installed-wheel check in CI when changing this boundary. Keep doctor instructions +agent-scoped and reuse `amplifier-config` rather than duplicate its workflow. diff --git a/README.md b/README.md index c604f624..3320f569 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,48 @@ saved selections and user-added bundles are unchanged. ## Commands +### Passive CLI diagnosis + +```bash +amplifier doctor # Local runtime and settings declarations +amplifier doctor --format json # One structured report +amplifier doctor --session # Include this project's session settings +``` + +Doctor reads installed distribution metadata and bounded settings files from the +current directory and Amplifier home. It does not resolve effective settings, +fetch/install dependencies, load credentials, mount modules, or repair anything. +Existing-session state remains **unknown**, including when `--session` is used. +Exit codes: `0` collection complete, `1` partial/refused, `2` invalid arguments. +Collection complete is not a health verdict. +The JSON report always marks health `not_assessed`; `issues` records collection +limits, not every possible environment defect. +The bounded diagnostic YAML parser intentionally accepts less syntax than the +runtime parser: aliases, explicit tags, and duplicate keys report +`unsupported_yaml`, not malformed `invalid_yaml`. This is a collection limit, +not an instruction to repair settings that the runtime accepts. + +Declaration values and arbitrary names use report-local equality refs. Scope, +container/member and zero-based parsed-order indices locate their declarations; +they do not establish precedence or serve as mutation selectors. Local paths +still identify the environment: review a report before sharing. + +The installed console script and `python -m amplifier_app_cli doctor` take a +passive startup route, including launchers targeting the legacy +`amplifier_app_cli.main:main` entry. Python interpreter startup and its initial +import-cache handling are outside that guarantee; later application bytecode +is suppressed. +The passive route requires `doctor` as the exact first argument after the +launcher; preceding arguments do not take this route. +The lightweight console dispatcher avoids importing the full interactive CLI +implementation on that route. Explicit Python access to the public `cli` +export still initializes the ordinary Click group, regardless of process arguments. + +For agentic interpretation, delegate to **`app-cli:cli-doctor`** in a CLI session. +It uses the report and loads the existing `amplifier-config` skill when needed. +The agent's own child state is not proof of another session's live state. +Diagnosis does not authorize changes or live probes. + ### Configuration Commands ```bash diff --git a/agents/cli-doctor.md b/agents/cli-doctor.md new file mode 100644 index 00000000..a4affaac --- /dev/null +++ b/agents/cli-doctor.md @@ -0,0 +1,102 @@ +--- +meta: + name: cli-doctor + description: >- + USE WHEN an Amplifier CLI environment needs diagnosis: runtime/source + identity, settings declarations, or missing capabilities. Returns collection + evidence, never a health or repair verdict; preserve that limit when + summarizing. DO NOT USE for project bugs or other hosts. For command usage, + use app-cli:cli-expert; for changes, consult amplifier-config. +model_role: general +--- + +# Amplifier CLI Doctor + +Diagnose the CLI application, not the user's project or another host. Default +to inspection. A diagnosis request does not authorize repair. + +## 1. Identify the target + +Establish the intended CLI executable, invocation directory, and any affected +session. Use target information already supplied; ask only for missing +information that changes the diagnosis. + +You run in a child session. Your interpreter, working directory and mounted +tools do not automatically describe your parent or another session. Attribute +each observation to its actual target; otherwise mark it unknown. + +**Success criteria:** the target and the limits of your access are explicit. + +## 2. Collect passive evidence + +From the intended directory and selected installed CLI, invoke: + +```text +amplifier doctor --format json +``` + +Use `--session SESSION_ID` only for an exact, supplied session in that project. +It selects the session's settings file; it neither resumes it nor observes its +live state. Do not guess a session ID or search private transcripts. + +This command reports installed distribution metadata and bounded, per-scope +declarations. It does not resolve effective settings, fetch sources, mount +modules or prove live behavior. Exit 0 means collection completed, not healthy; +exit 1 means partial collection or refusal; exit 2 means invalid arguments. +Retain the actual command, tool result and issue codes, not just a summary. +The report's `health.status` is `not_assessed`. An empty `issues` list means no +collection problems were recorded, not that nothing needs fixing. + +Older installed CLIs may not have this command. If unavailable, report that +limit. Do not install/update anything or substitute inspection-looking commands +that can activate dependencies. Consult `app-cli:cli-expert` for version-matched +command semantics when needed and available. + +**Success criteria:** genuine target-command evidence or a concrete collection +limit, without environment changes. + +## 3. Interpret without inventing runtime truth + +Keep declarations, effective settings, installed code, mounted objects and +successful workflows separate. Equal opaque refs mean equality only within one +report. Locate declarations by scope, fixed container/member, and zero-based +parsed-order index. These are inspection locators, not safe mutation selectors. +Re-read and confirm any target before a later authorized edit. + +For configuration analysis, load the existing `amplifier-config` skill with +`load_skill`. Confirm its returned source; report missing or shadowed guidance +rather than inventing a replacement. Reuse its configuration workflow instead +of reproducing merge, routing, skill-discovery or provider semantics here. +Do not load it for an identity-only request that does not need it. + +Never expand or disclose credential values. The passive report withholds +arbitrary declaration names and source URLs, but local paths still identify +the environment. Review before sharing; do not upload reports automatically. + +**Success criteria:** each conclusion has supporting evidence and unresolved +effective/live state remains unknown. + +## 4. Return a bounded next step + +Report findings, evidence, unknowns, the smallest supported next action and how +to verify it. A clean passive report does not establish repair or workflow +success. Root success does not certify a child. +State **"Health not assessed; no repair verdict from passive collection."** +Do not conclude "no repair needed", "nothing needs fixing" or "healthy" from +`collection_status: complete` or `issues: []`. If the question requires effective +or live state, name the missing observation instead of turning it into a pass. +Doctor always leaves effective and live state unknown, even after a repair or +with `--session`. Never use a changed `effective_settings.status` as its success +criterion. Verify an actual affected setting or behavior through a separately +supported, authorized observation. If no concrete symptom or failed behavior +was supplied, the next step is to identify that symptom, not to certify that no +repair is needed or invent another effective-settings command. + +Do not reset caches, reinstall, switch roots/providers, alter permissions, +disable modes, edit guidance, load credentials, replay work or start live +probes under diagnosis authority. Propose those separately with scope, +activation effects and rollback; configuration changes must follow the loaded +skill's confirmation rules. Never bypass an ownership or access refusal. + +**Success criteria:** an actionable diagnosis, not an unsupported health grade +or an unrequested repair. \ No newline at end of file diff --git a/amplifier_app_cli/__init__.py b/amplifier_app_cli/__init__.py index 27056fa3..2cca5038 100644 --- a/amplifier_app_cli/__init__.py +++ b/amplifier_app_cli/__init__.py @@ -1,6 +1,17 @@ """Amplifier CLI - Command-line interface for the Amplifier platform.""" +import sys + +# Only the exact public doctor route bypasses normal application startup. +# Python may already have cached this first __init__; suppress later bytecode. +if len(sys.argv) > 1 and sys.argv[1] == "doctor": + sys.dont_write_bytecode = True + +import importlib +from types import ModuleType as _ModuleType + __version__ = "0.1.0" +__all__ = ["cli", "main", "__version__"] # TEMPORARY, remove when truststore ships the fix. This must run before any # TLS happens in this process, and nothing in the amplifier tree owns the @@ -8,11 +19,42 @@ # package import is the earliest amplifier-owned point available. It is a # no-op unless the installed truststore's wrap_bio is genuinely unlocked. # See truststore_shim.py for the full story. -from . import truststore_shim as _truststore_shim +def _ensure_tls_shim(): + from . import truststore_shim as _truststore_shim -_truststore_shim.apply() + _truststore_shim.apply() -from .main import cli -from .main import main -__all__ = ["cli", "main", "__version__"] +def __getattr__(name): + if name not in __all__: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + module = importlib.import_module(".main", __name__) + if name == "cli": + module._initialize_cli() + globals()["cli"] = module.cli + # Importing the submodule sets package.main to that module. Restore the + # public dispatcher, including when the submodule was imported first. + globals()["main"] = module.main + return globals()[name] + + +if not (len(sys.argv) > 1 and sys.argv[1] == "doctor"): + _ensure_tls_shim() + __getattr__("cli") + + +class _PackageModule(_ModuleType): + def __getattribute__(self, name): + value = super().__getattribute__(name) + # Python publishes imported children on their parent, bypassing + # __getattr__. Resolve only the collision with our public dispatcher. + if ( + name == "main" + and isinstance(value, _ModuleType) + and value is sys.modules.get(f"{__name__}.main") + ): + return __getattr__("main") + return value + + +sys.modules[__name__].__class__ = _PackageModule diff --git a/amplifier_app_cli/__main__.py b/amplifier_app_cli/__main__.py index 7df2a191..64588d8e 100644 --- a/amplifier_app_cli/__main__.py +++ b/amplifier_app_cli/__main__.py @@ -1,6 +1,20 @@ -"""Entry point for running as module.""" +"""Published console and module entry point.""" + +import sys + + +def main(): + """Dispatch doctor before importing the normal application's runtime.""" + if len(sys.argv) > 1 and sys.argv[1] == "doctor": + # The package can already be cached from an earlier normal invocation. + sys.dont_write_bytecode = True + from .commands.doctor import main as doctor_main + + return doctor_main() + from .main import main as application_main + + return application_main() -from .main import main if __name__ == "__main__": main() diff --git a/amplifier_app_cli/commands/__init__.py b/amplifier_app_cli/commands/__init__.py index 47964c98..3b563af6 100644 --- a/amplifier_app_cli/commands/__init__.py +++ b/amplifier_app_cli/commands/__init__.py @@ -1,15 +1,7 @@ """CLI command exports for amplifier-app-cli.""" -from .init import auto_init_from_env -from .init import check_first_run -from .init import init_cmd -from .init import prompt_first_run_init -from .module import module -from .provider import provider -from .run import register_run_command -from .session import register_session_commands -from .source import source -from .tool import tool +import sys +from types import ModuleType as _ModuleType __all__ = [ "auto_init_from_env", @@ -23,3 +15,58 @@ "source", "tool", ] +_exports_loaded = False + + +def _load_exports(): + global _exports_loaded + global auto_init_from_env, check_first_run, init_cmd, prompt_first_run_init + global module, provider, register_run_command, register_session_commands, source, tool + if _exports_loaded: + return + from .. import _ensure_tls_shim + + _ensure_tls_shim() + # Bind the complete fixed public surface together: importing a submodule + # otherwise leaves same-named attributes such as provider as module objects. + from .init import auto_init_from_env + from .init import check_first_run + from .init import init_cmd + from .init import prompt_first_run_init + from .module import module + from .provider import provider + from .run import register_run_command + from .session import register_session_commands + from .source import source + from .tool import tool + + _exports_loaded = True + + +def __getattr__(name): + if name not in __all__: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + _load_exports() + return globals()[name] + + +if not (len(sys.argv) > 1 and sys.argv[1] == "doctor"): + _load_exports() + + +class _PackageModule(_ModuleType): + def __getattribute__(self, name): + value = super().__getattribute__(name) + # A directly imported child can mask its same-named public command; + # __getattr__ alone cannot see an attribute that Python already set. + if ( + name in ("module", "provider", "source", "tool") + and isinstance(value, _ModuleType) + and value is sys.modules.get(f"{__name__}.{name}") + ): + _load_exports() + return getattr(value, name) + return value + + +sys.modules[__name__].__class__ = _PackageModule diff --git a/amplifier_app_cli/commands/doctor.py b/amplifier_app_cli/commands/doctor.py new file mode 100644 index 00000000..5cecbae7 --- /dev/null +++ b/amplifier_app_cli/commands/doctor.py @@ -0,0 +1,59 @@ +"""Thin passive diagnostic command. No normal CLI startup imports.""" + +import json +import sys + +import click + +from ..lib.diagnostics import collect_diagnostics +from ..lib.diagnostics import current_target +from ..lib.diagnostics import render_text +from ..lib.diagnostics import valid_session_id + + +def _session(_ctx, _param, value): + if value is not None and not valid_session_id(value): + raise click.BadParameter("invalid_session_id") + return value + + +class _DoctorCommand(click.Command): + def parse_args(self, ctx, args): + try: + return super().parse_args(ctx, args) + except click.ClickException: + raise click.UsageError("invalid_doctor_arguments", ctx) from None + + +@click.command(cls=_DoctorCommand) +@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text") +@click.option("--session", "session_id", callback=_session, metavar="SESSION_ID") +def doctor(output_format, session_id): + """Report local declarations without resolving or activating them.""" + try: + target = current_target(session_id) + report = collect_diagnostics(target) + except Exception: + # Never expose path, parser, metadata, or OS exception payloads. + from ..lib.diagnostics import unavailable_report + + report = unavailable_report() + click.echo(json.dumps(report, ensure_ascii=True) if output_format == "json" else render_text(report)) + if report["collection_status"] != "complete": + raise click.exceptions.Exit(1) + + +def main(): + """Exact-route entry; malformed doctor input cannot fall back to main.""" + try: + code = doctor.main(args=sys.argv[2:], prog_name="amplifier doctor", standalone_mode=False) + raise SystemExit(code or 0) + except click.ClickException: + click.echo("Usage: amplifier doctor [--format text|json] [--session SESSION_ID]", err=True) + click.echo("Error: invalid_doctor_arguments", err=True) + raise SystemExit(2) from None + except click.exceptions.Exit as error: + raise SystemExit(error.exit_code) from None + except (EOFError, KeyboardInterrupt): + click.echo("Error: doctor_interrupted", err=True) + raise SystemExit(1) from None \ No newline at end of file diff --git a/amplifier_app_cli/lib/diagnostics.py b/amplifier_app_cli/lib/diagnostics.py new file mode 100644 index 00000000..b6a6b205 --- /dev/null +++ b/amplifier_app_cli/lib/diagnostics.py @@ -0,0 +1,515 @@ +"""Bounded, passive CLI provenance collection. + +This is not a settings resolver or a health verdict. Only allowlisted facts +leave this module. Arbitrary declarations receive report-local equality refs, +never display names, URLs, or values. Paths describe local identity and are not +a promise that a report is suitable for public sharing. +""" + +from __future__ import annotations + +import importlib.metadata as metadata +import json +import os +from pathlib import Path +import re +import stat +import sys +import unicodedata +from dataclasses import dataclass +from urllib.parse import urlsplit +from urllib.request import url2pathname + +import yaml + +from ..project_utils import get_project_slug +from .venv_home_guard import SharedVenvHomeError, enforce_home_ownership + +SETTINGS_LIMIT = 256 * 1024 +METADATA_LIMIT = 32 * 1024 +ENTRY_LIMIT = 200 +_ABSENT = object() +_SESSION = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,199}", re.ASCII) +# Do not publish arbitrary local build labels from untrusted metadata. +_VERSION = re.compile( + r"(?:[0-9]+!)?[0-9]+(?:\.[0-9]+)*(?:(?:a|b|rc)[0-9]+)?" + r"(?:\.post[0-9]+)?(?:\.dev[0-9]+)?", + re.ASCII, +) + + +@dataclass(frozen=True) +class DiagnosticTarget: + cwd: Path + home: Path + session_id: str | None = None + + +def valid_session_id(value: str) -> bool: + return isinstance(value, str) and _SESSION.fullmatch(value) is not None + + +def current_target(session_id: str | None = None) -> DiagnosticTarget: + """Resolve the same home/layout as settings, without importing Foundation.""" + if session_id is not None and not valid_session_id(session_id): + raise ValueError("invalid_session_id") + home = os.environ.get("AMPLIFIER_HOME") + return DiagnosticTarget( + Path.cwd().resolve(), + Path(home).expanduser().resolve() if home else (Path.home() / ".amplifier").resolve(), + session_id, + ) + + +def _safe_path(value, issues): + text = str(value) + # Refuse terminal controls, bidi/format controls and URI/credential forms. + if ( + len(text) > 4096 + or any(unicodedata.category(c).startswith("C") for c in text) + or re.search(r"[A-Za-z][A-Za-z0-9+.-]*://|[?&#@]|%[0-9A-Fa-f]{2}", text) + or text.startswith(("//", "\\\\")) + ): + issues("path_withheld") + return None + return text + + +def _base_report(): + return { + "schema_version": 1, + "mode": "passive", + "collection_status": "refused", + "health": {"status": "not_assessed", "reason": "collection_only"}, + "target": {"cwd": None, "home": None, "session_selected": False}, + "ownership": {"status": "unknown", "foreign_count": None, "owned_count": None}, + "runtime": None, + "scopes": [ + {"scope": scope, "path": None, "status": "not_selected"} + for scope in ("global", "project", "local", "session") + ], + "declarations": [], + "environment_channels": {"module_count": 0, "modules": []}, + "effective_settings": {"status": "unknown", "reason": "effective_resolution_not_performed"}, + "live_session": {"status": "unknown", "reason": "not_observed"}, + "issues": [], + } + + +def unavailable_report(): + report = _base_report() + report["issues"] = [{"code": "target_unavailable"}] + return report + + +class _LimitError(ValueError): + pass + + +class UnsupportedYamlError(ValueError): + """Syntax outside the passive diagnostic parser's accepted subset.""" + + +class _DiagnosticLoader(yaml.SafeLoader): + """SafeLoader with diagnostic-only resource and syntax restrictions.""" + + def __init__(self, stream): + super().__init__(stream) + self.depth = 0 + self.nodes = 0 + + def compose_node(self, parent, index): + event = self.peek_event() + if isinstance(event, yaml.AliasEvent) or getattr(event, "tag", None) is not None: + raise UnsupportedYamlError() + self.depth += 1 + self.nodes += 1 + if self.depth > 32 or self.nodes > 10000: + raise _LimitError() + if isinstance(event, yaml.ScalarEvent) and len(event.value.encode("utf-8")) > 16384: + raise _LimitError() + try: + return super().compose_node(parent, index) + finally: + self.depth -= 1 + + def construct_mapping(self, node, deep=False): + keys = set() + for key_node, _ in node.value: + key = self.construct_object(key_node, deep=deep) + if key in keys: + raise UnsupportedYamlError() + keys.add(key) + return super().construct_mapping(node, deep=deep) + + +def _read_regular(path: Path, limit: int): + """Check before opening; nonblocking/nofollow also protect common races.""" + try: + mode = path.lstat().st_mode + if stat.S_ISLNK(mode): + return "unsupported_symlink", None + if not stat.S_ISREG(mode): + return "unsupported_file", None + flags = os.O_RDONLY | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0) + fd = os.open(path, flags) + with os.fdopen(fd, "rb") as stream: + info = os.fstat(stream.fileno()) + if not stat.S_ISREG(info.st_mode): + return "unsupported_file", None + if info.st_size > limit: + return "too_large", None + data = stream.read(limit + 1) + if len(data) > limit: + return "too_large", None + return "valid", data + except FileNotFoundError: + return "missing", None + except (OSError, ValueError): + return "unreadable", None + + +def _read_settings(path): + status, data = _read_regular(path, SETTINGS_LIMIT) + if status != "valid": + return status, None + try: + loader = _DiagnosticLoader(data.decode("utf-8")) + try: + parsed = loader.get_single_data() + finally: + loader.dispose() + if parsed is None: + return "empty", None + if not isinstance(parsed, dict): + return "invalid_shape", parsed + return ("empty" if not parsed else "valid"), parsed + except _LimitError: + return "too_large", None + except UnsupportedYamlError: + return "unsupported_yaml", None + except Exception: + return "invalid_yaml", None + + +def _shape(value): + if value is _ABSENT: + return "absent" + if value is None: + return "null" + if value is False: + return "false" + if value is True: + return "true" + if isinstance(value, str): + return "string" if value else "empty_string" + if isinstance(value, dict): + return "mapping" if value else "empty_mapping" + if isinstance(value, list): + return "list" if value else "empty_list" + if isinstance(value, int): + return "integer" + if isinstance(value, float): + return "number" + return "other_scalar" + + +def _source_kind(value): + if not isinstance(value, str): + return "not_string" + if value.startswith(("git+", "git://", "git@")): + return "git" + if value.startswith("file:"): + return "file" + if value.startswith(("http:", "https:")): + return "http" + if value.startswith(("/", ".", "~", "\\")) or re.match(r"^[A-Za-z]:", value): + return "path" + return "opaque" + + +def _declarations(data, scope, emit, ref): + """Fixed fields and parsed-order indices are inspection locators, not selectors.""" + def get(parent, key): + return parent.get(key, _ABSENT) if isinstance(parent, dict) else _ABSENT + + def field(container, member, value, **extra): + row = {"scope": scope, "container": container, "member": member, "shape": _shape(value), **extra} + if value is not _ABSENT and not isinstance(value, (dict, list)): + row["value_ref"] = ref(value) + row["source_kind"] = _source_kind(value) + emit(row) + + for container, member in ( + ("bundle", "active"), ("bundle", "app"), ("routing", "matrix"), + ("sources", "modules"), ("sources", "bundles"), + ("overrides", "modules"), ("overrides", "bundles"), + ): + parent = data + for key in container.split("."): + parent = get(parent, key) + value = get(parent, member) + field(container, member, value, container_shape=_shape(parent)) + if container == "bundle" and member == "app" and isinstance(value, list): + for index, item in enumerate(value): + field("bundle.app", "item", item, index=index) + if container == "sources" and isinstance(value, dict): + for index, (identity, item) in enumerate(value.items()): + field("sources." + member, "source", item, index=index, identity_ref=ref(identity)) + + overrides = get(data, "overrides") + field("overrides", "entries", overrides) + if isinstance(overrides, dict): + for index, (identity, item) in enumerate(overrides.items()): + if identity in ("modules", "bundles"): + # Development maps: report presence/shape only, not their targets. + continue + field("overrides", "source", get(item, "source"), index=index, identity_ref=ref(identity)) + + for container in ("modules", "config"): + parent = get(data, container) + providers = get(parent, "providers") + field(container, "providers", providers, container_shape=_shape(parent)) + if isinstance(providers, list): + for index, item in enumerate(providers): + extra = {} + for key in ("id", "module"): + identity = get(item, key) + if identity is not _ABSENT: + extra[key + "_ref"] = ref(identity) + field( + container + ".providers", "source", get(item, "source"), + index=index, entry_shape=_shape(item), **extra, + ) + + +def _direct_url(dist, issue): + result = {"status": "missing"} + try: + # Use metadata APIs only: never import or load installed entry points. + entry = next( + (p for p in (dist.files or ()) if len(p.parts) == 2 + and p.parts[0].endswith(".dist-info") and p.parts[1] == "direct_url.json"), + None, + ) + if entry is None: + return result + status, data = _read_regular(Path(dist.locate_file(entry)), METADATA_LIMIT) + if status != "valid": + issue("distribution_metadata_unavailable") + return {"status": status} + parsed = json.loads(data) + if not isinstance(parsed, dict): + raise ValueError() + result = {"status": "valid", "editable": None, "local_target": None, "commit": None} + directory = parsed.get("dir_info") + if isinstance(directory, dict) and type(directory.get("editable")) is bool: + result["editable"] = directory["editable"] + url = parsed.get("url") + if isinstance(url, str): + if any(unicodedata.category(c).startswith("C") for c in url): + raise ValueError() + parts = urlsplit(url) + if parts.scheme == "file" and not parts.netloc and not parts.query and not parts.fragment: + # url2pathname performs exactly one percent decode. + native = url2pathname(parts.path) + if Path(native).is_absolute(): + result["local_target"] = _safe_path(native, issue) + vcs = parsed.get("vcs_info") + commit = vcs.get("commit_id") if isinstance(vcs, dict) else None + if isinstance(commit, str) and re.fullmatch(r"[a-fA-F0-9]{7,64}", commit): + result["commit"] = commit + return result + except Exception: + issue("distribution_metadata_unavailable") + return {"status": "unreadable"} + + +def _runtime(issue): + distributions = [] + for name in ("amplifier", "amplifier-app-cli", "amplifier-core", "amplifier-foundation"): + row = {"name": name, "status": "missing"} + try: + dist = metadata.distribution(name) + version = dist.version + row = { + "name": name, "status": "present", + "version": version if isinstance(version, str) and len(version) <= 128 and _VERSION.fullmatch(version) else None, + "location": _safe_path(dist.locate_file(""), issue), + "direct_url": _direct_url(dist, issue), + } + if row["version"] is None: + issue("distribution_version_withheld") + except metadata.PackageNotFoundError: + pass + except Exception: + row["status"] = "unreadable" + issue("distribution_metadata_unavailable") + distributions.append(row) + return { + "argv0": _safe_path(sys.argv[0], issue), + "interpreter": _safe_path(sys.executable, issue), + "python": ".".join(str(n) for n in sys.version_info[:3]), + "prefix": _safe_path(sys.prefix, issue), + "base_prefix": _safe_path(sys.base_prefix, issue), + "distributions": distributions, + } + + +def collect_diagnostics(target: DiagnosticTarget) -> dict: + """Collect bounded local facts. Complete means collected, not healthy.""" + report = _base_report() + + def issue(code, scope=None): + row = {"code": code} + if scope is not None: + row["scope"] = scope + if row not in report["issues"]: + report["issues"].append(row) + + report["target"] = { + "cwd": _safe_path(target.cwd, issue), + "home": _safe_path(target.home, issue), + "session_selected": target.session_id is not None, + } + if target.session_id is not None and not valid_session_id(target.session_id): + issue("invalid_session_id") + return report + try: + conflict = enforce_home_ownership(["doctor"], current_home=target.home) + report["ownership"] = { + "status": "override" if conflict is not None else "clear", + "foreign_count": len(conflict.foreign) if conflict is not None else 0, + "owned_count": len(conflict.owned) if conflict is not None else None, + } + except SharedVenvHomeError as error: + report["ownership"] = { + "status": "refused", "foreign_count": len(error.conflict.foreign), + "owned_count": len(error.conflict.owned), + } + issue("ownership_conflict") + return report + except Exception: + issue("ownership_unknown") + return report + + report["runtime"] = _runtime(issue) + refs = {} + + def ref(value): + # YAML values never leave this local dictionary. Typed repr also + # distinguishes null, false, empty, and numeric/string identities. + key = (type(value).__name__, repr(value)) + if key not in refs: + refs[key] = f"r{len(refs) + 1}" + return refs[key] + + def emit(row): + if len(report["declarations"]) < ENTRY_LIMIT: + report["declarations"].append(row) + else: + issue("declarations_truncated") + + paths = [ + target.home / "settings.yaml", + target.cwd / ".amplifier" / "settings.yaml", + target.cwd / ".amplifier" / "settings.local.yaml", + None, + ] + if target.session_id is not None: + # get_project_slug is a pure cwd helper; explicit targets need the + # same transform without changing process cwd. + slug = get_project_slug() if target.cwd == Path.cwd().resolve() else ( + str(target.cwd.resolve()).replace("/", "-").replace("\\", "-").replace(":", "") + ) + if not slug.startswith("-"): + slug = "-" + slug + paths[3] = target.home / "projects" / slug / "sessions" / target.session_id / "settings.yaml" + for row, path in zip(report["scopes"], paths): + if path is None: + continue + scope = row["scope"] + row["path"] = _safe_path(path, issue) + try: + if scope == "session" and any(p.is_symlink() for p in (path, *path.parents)): + status, data = "unsupported_symlink", None + else: + status, data = _read_settings(path) + except Exception: + status, data = "unreadable", None + row["status"] = status + if status in ("valid", "empty", "invalid_shape"): + row["shape"] = _shape(data) + if status not in ("valid", "empty", "missing"): + issue("settings_" + status, scope) + if status in ("valid", "empty"): + _declarations(data or {}, scope, emit, ref) + + channels = report["environment_channels"] + for name in ("AMPLIFIER_GIT_HOST", "AMPLIFIER_HOME_ENV"): + channels[name] = {"present": name in os.environ} + for name in os.environ: + if name.startswith("AMPLIFIER_MODULE_"): + channels["module_count"] += 1 + if len(channels["modules"]) < ENTRY_LIMIT: + channels["modules"].append({"identity_ref": ref(name), "present": True}) + else: + issue("environment_channels_truncated") + report["collection_status"] = "partial" if report["issues"] else "complete" + return report + + +def render_text(report: dict) -> str: + """Render the allowlisted report, with no resolution or health inference.""" + lines = [ + "Amplifier doctor (passive)", + "Collection: " + report["collection_status"] + " (not a health verdict)", + "Health: not assessed. Issues describe collection limits, not all environment defects.", + "Paths identify this local environment; review before sharing.", + "CWD: " + (report["target"]["cwd"] or "withheld"), + "Home: " + (report["target"]["home"] or "withheld"), + "Ownership: " + report["ownership"]["status"], + ] + runtime = report["runtime"] + if runtime is not None: + lines.extend(( + "Interpreter: " + (runtime["interpreter"] or "withheld"), + "Launcher: " + (runtime["argv0"] or "withheld"), + "Python: " + runtime["python"], + "Prefix: " + (runtime["prefix"] or "withheld"), + "Base prefix: " + (runtime["base_prefix"] or "withheld"), + )) + for row in runtime["distributions"]: + lines.append(row["name"] + ": " + row["status"] + " " + (row.get("version") or "")) + if row.get("location") is not None: + lines.append(" Location: " + row["location"]) + direct = row.get("direct_url", {}) + if direct.get("status") == "valid": + editable = direct["editable"] + lines.append(" Editable: " + ("unknown" if editable is None else str(editable).lower())) + if direct["local_target"] is not None: + lines.append(" Local target: " + direct["local_target"]) + if direct["commit"] is not None: + lines.append(" VCS commit: " + direct["commit"]) + for row in report["scopes"]: + lines.append(row["scope"] + ": " + row["status"] + " " + (row["path"] or "")) + for row in report["declarations"]: + index = f"[{row['index']}]" if "index" in row else "" + refs = " ".join(f"{key}={row[key]}" for key in ("identity_ref", "id_ref", "module_ref", "value_ref") if key in row) + kind = " kind=" + row["source_kind"] if "source_kind" in row else "" + lines.append(f"{row['scope']} {row['container']}{index}.{row['member']}: {row['shape']}{kind} {refs}") + channels = report["environment_channels"] + lines.append(f"Environment module channels: {channels['module_count']} (values not read)") + for row in channels["modules"]: + lines.append(" Module channel: " + row["identity_ref"] + " present") + for name in ("AMPLIFIER_GIT_HOST", "AMPLIFIER_HOME_ENV"): + if name in channels: + lines.append(name + ": " + ("present" if channels[name]["present"] else "absent")) + lines.extend(( + "Effective settings: unknown (effective_resolution_not_performed)", + "Live session: unknown (not_observed)", + "Refs express equality only within this report; no precedence or mounted-state claims.", + )) + for issue in report["issues"]: + lines.append("Issue: " + issue["code"] + (" (" + issue["scope"] + ")" if "scope" in issue else "")) + return "\n".join(lines) \ No newline at end of file diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index 1ccc431f..8cd88319 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -1,5 +1,7 @@ """Amplifier CLI - Command-line interface for the Amplifier platform.""" +from __future__ import annotations + import asyncio import html import json @@ -12,104 +14,15 @@ from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path +from threading import RLock from typing import TYPE_CHECKING, Any -import click - -from amplifier_app_cli.utils.help_formatter import AmplifierGroup - if TYPE_CHECKING: from amplifier_foundation.bundle import PreparedBundle -from amplifier_core import ( - AmplifierSession, - ModuleValidationError, # pyright: ignore[reportAttributeAccessIssue] -) -from amplifier_core.llm_errors import LLMError -from amplifier_foundation import sanitize_message -from prompt_toolkit import PromptSession -from prompt_toolkit.buffer import Buffer -from prompt_toolkit.completion import CompleteEvent -from prompt_toolkit.filters import has_completions -from prompt_toolkit.formatted_text import HTML -from prompt_toolkit.history import FileHistory, InMemoryHistory -from prompt_toolkit.key_binding import KeyBindings -from prompt_toolkit.shortcuts import CompleteStyle -from rich.panel import Panel - -# Errors that mean the terminal can never satisfy the REPL, so retrying is -# pointless. See the REPL loop's handler for the full story: on Windows without -# a real console, prompt_toolkit's Win32Output raises before the loop reaches -# any await, so the "keep going" catch-all turns into an unkillable busy spin. -# -# `prompt_toolkit.output.win32` asserts `sys.platform == "win32"` at import, so -# this must stay guarded. On POSIX the tuple is empty and `except ()` catches -# nothing -- the POSIX path is byte-identical in effect to what shipped. -if sys.platform == "win32": # pragma: no cover - platform-specific - from prompt_toolkit.output.win32 import NoConsoleScreenBufferError - - _TERMINAL_UNUSABLE_ERRORS: tuple[type[BaseException], ...] = ( - NoConsoleScreenBufferError, - ) -else: - _TERMINAL_UNUSABLE_ERRORS = () - - -from amplifier_foundation.paths.resolution import get_amplifier_home -from .commands.agents import agents as agents_group -from .commands.allowed_dirs import allowed_dirs as allowed_dirs_group -from .commands.bundle import bundle as bundle_group -from .commands.denied_dirs import denied_dirs as denied_dirs_group -from .commands.init import check_first_run, init_cmd, prompt_first_run_init -from .commands.module import module as module_group -from .commands.notify import notify as notify_group -from .commands.provider import provider as provider_group -from .commands.reset import reset as reset_cmd -from .commands.routing import routing_group -from .commands.run import register_run_command -from .commands.session import register_session_commands -from .commands.source import source as source_group -from .commands.tool import tool as tool_group -from .commands.update import update as update_cmd -from .commands.version import version as version_cmd -from .console import Markdown, console -from .dedicated_tty_input import close_dedicated_tty_input, get_dedicated_tty_input -from .effective_config import get_effective_config_summary -from .key_manager import KeyManager -from .lib.settings import AppSettings -from .provider_diagnostics import DEFAULT_TIMEOUT_S as _PROVIDER_DIAGNOSTIC_TIMEOUT_S -from .provider_diagnostics import format_model_line -from .provider_diagnostics import invoke_list_models -from .provider_diagnostics import test_provider_connectivity -from .session_runner import ( - SessionConfig, - create_initialized_session, - repair_transcript_if_needed, -) -from .session_store import SessionStore -from .stdout_offload import patch_stdout_offloaded as patch_stdout -from .ui.dashboard_renderer import DashboardRenderer -from .ui.dashboard_renderer import _redact_value as _dr_redact_value -from .ui.error_display import display_llm_error, display_validation_error -from .ui.item_renderer import ItemRenderer -from .ui.log_filter import LLMErrorLogFilter -from .ui.completion import ( - Candidate, - SlashCompleter, - build_completion_snapshot, - longest_common_prefix, -) -from .ui.view_policy import resolve_view -from .utils.error_format import escape_markup -from .utils.shell_completion import ( - can_safely_modify as _completion_can_safely_modify, - detect_shell as _completion_detect_shell, - get_shell_config_file as _completion_config_file, - install_completion_to_config as _install_completion, - is_click_completion_instruction, - is_completion_installed as _completion_is_installed, - manual_installation_instruction, -) -from .utils.version import get_core_version, get_version + +_cli_ready = False +_cli_initializing = False +_cli_lock = RLock() def _report_terminal_unusable(exc: BaseException, *, verbose: bool = False) -> None: @@ -121,8 +34,7 @@ def _report_terminal_unusable(exc: BaseException, *, verbose: bool = False) -> N prompt_toolkit traceback naming ``Win32Output`` -- accurate, but it points at a library internal rather than at what they did or what to do instead. """ - # Local imports: this helper is defined above main.py's own import block so - # it can sit next to the _TERMINAL_UNUSABLE_ERRORS tuple it belongs to. + # Keep the terminal-error reporting usable independently of CLI startup. from .console import console as _console from .utils.error_format import escape_markup as _escape @@ -141,8 +53,6 @@ def _report_terminal_unusable(exc: BaseException, *, verbose: bool = False) -> N _console.print_exception() -logger = logging.getLogger(__name__) - # --------------------------------------------------------------------------- # Cleanup-window observability events # @@ -163,16 +73,6 @@ def _report_terminal_unusable(exc: BaseException, *, verbose: bool = False) -> N CLEANUP_FINALLY_BEGIN: str = "cleanup:finally_begin" CLEANUP_FINALLY_END: str = "cleanup:finally_end" -# Suppress duplicate LLM error lines from console output. -# The CLI renders LLM errors as Rich panels — the logger.error() calls -# from the provider ("[PROVIDER] Anthropic API error: ...") and session -# ("Execution failed: ...") would duplicate the same info as raw text. -# This filter must be attached to the console HANDLER (not the root logger) -# so it intercepts records propagated from child loggers like provider modules. -# Log file handlers managed by hooks use their own loggers and are unaffected. -_llm_error_filter = LLMErrorLogFilter() - - def _ensure_utf8_output() -> None: """Force UTF-8 on stdout/stderr so terminal-rendered Unicode survives copy/paste. @@ -291,12 +191,6 @@ def _attach_llm_error_filter() -> None: root.addFilter(_llm_error_filter) -# Load API keys from ~/.amplifier/keys.env on startup -# This allows keys saved by 'amplifier init' or 'amplifier provider use' to be available -if not os.environ.get("_AMPLIFIER_COMPLETE"): - KeyManager() - - # Placeholder for the run command; assigned after registration below _run_command: Callable | None = None @@ -3264,20 +3158,7 @@ def get_module_search_paths() -> list[Path]: return paths -@click.group(cls=AmplifierGroup, invoke_without_command=True) -@click.version_option( - version=f"{get_version()} (core {get_core_version()})", - prog_name="amplifier", -) -@click.option( - "--install-completion", - is_flag=True, - flag_value="auto", - default=None, - help="Install completion for the shell detected from $SHELL (bash, zsh, or fish)", -) -@click.pass_context -def cli(ctx, install_completion): +def _cli_callback(ctx, install_completion): """Amplifier - AI-powered modular development platform.""" # Handle --install-completion flag if install_completion: @@ -5005,42 +4886,202 @@ async def after_cleanup(): console.file = original_console_file -# Register standalone commands -cli.add_command(agents_group) -cli.add_command(allowed_dirs_group) -cli.add_command(denied_dirs_group) -cli.add_command(bundle_group) -cli.add_command(init_cmd) -cli.add_command(module_group) -cli.add_command(notify_group) -cli.add_command(provider_group) -cli.add_command(routing_group) -cli.add_command(source_group) -cli.add_command(tool_group) -cli.add_command(update_cmd) -cli.add_command(version_cmd) -cli.add_command(reset_cmd) - - -# Note: The *_with_session variants were removed in favor of unified functions -# that accept optional initial_transcript parameter for resume functionality. -# See execute_single() and interactive_chat() above. - -_run_command = register_run_command( - cli, - interactive_chat=interactive_chat, - execute_single=execute_single, - get_module_search_paths=get_module_search_paths, - check_first_run=check_first_run, - prompt_first_run_init=prompt_first_run_init, -) - -register_session_commands( - cli, - interactive_chat=interactive_chat, - execute_single=execute_single, - get_module_search_paths=get_module_search_paths, -) +def _initialize_cli() -> None: + """Build the ordinary CLI once, in this module's original global namespace. + + Legacy doctor imports defer this work, but explicit public ``cli`` access + must still return the real decorated group, regardless of process argv. + """ + global _cli_ready, _cli_initializing, cli, _run_command + global click, AmplifierGroup, AmplifierSession, ModuleValidationError, LLMError + global sanitize_message, PromptSession, Buffer, CompleteEvent, has_completions + global HTML, FileHistory, InMemoryHistory, KeyBindings, CompleteStyle, Panel + global NoConsoleScreenBufferError, _TERMINAL_UNUSABLE_ERRORS, get_amplifier_home + global agents_group, allowed_dirs_group, bundle_group, denied_dirs_group + global check_first_run, init_cmd, prompt_first_run_init, module_group, notify_group + global provider_group, reset_cmd, routing_group, register_run_command + global register_session_commands, source_group, tool_group, update_cmd, version_cmd + global Markdown, console, close_dedicated_tty_input, get_dedicated_tty_input + global get_effective_config_summary, KeyManager, AppSettings + global _PROVIDER_DIAGNOSTIC_TIMEOUT_S, format_model_line, invoke_list_models + global test_provider_connectivity, SessionConfig, create_initialized_session + global repair_transcript_if_needed, SessionStore, patch_stdout, DashboardRenderer + global _dr_redact_value, display_llm_error, display_validation_error, ItemRenderer + global LLMErrorLogFilter, Candidate, SlashCompleter, build_completion_snapshot + global longest_common_prefix, resolve_view, escape_markup + global _completion_can_safely_modify, _completion_detect_shell, _completion_config_file + global _install_completion, is_click_completion_instruction, _completion_is_installed + global manual_installation_instruction, get_core_version, get_version + global logger, _llm_error_filter + + with _cli_lock: + if _cli_ready: + return + if _cli_initializing: + raise RuntimeError("CLI initialization is already in progress") + _cli_initializing = True + try: + from . import _ensure_tls_shim + + _ensure_tls_shim() + import click + + from amplifier_app_cli.utils.help_formatter import AmplifierGroup + from amplifier_core import ( + AmplifierSession, + ModuleValidationError, # pyright: ignore[reportAttributeAccessIssue] + ) + from amplifier_core.llm_errors import LLMError + from amplifier_foundation import sanitize_message + from prompt_toolkit import PromptSession + from prompt_toolkit.buffer import Buffer + from prompt_toolkit.completion import CompleteEvent + from prompt_toolkit.filters import has_completions + from prompt_toolkit.formatted_text import HTML + from prompt_toolkit.history import FileHistory, InMemoryHistory + from prompt_toolkit.key_binding import KeyBindings + from prompt_toolkit.shortcuts import CompleteStyle + from rich.panel import Panel + + # On Windows without a real console, retrying the REPL would spin. + # This import asserts win32; on POSIX except () catches nothing. + if sys.platform == "win32": # pragma: no cover - platform-specific + from prompt_toolkit.output.win32 import NoConsoleScreenBufferError + + _TERMINAL_UNUSABLE_ERRORS = (NoConsoleScreenBufferError,) + else: + _TERMINAL_UNUSABLE_ERRORS = () + + from amplifier_foundation.paths.resolution import get_amplifier_home + from .commands import _load_exports + + # Preserve the original commands-package import ordering and bind + # same-named public command objects before importing submodules. + _load_exports() + from .commands.agents import agents as agents_group + from .commands.allowed_dirs import allowed_dirs as allowed_dirs_group + from .commands.bundle import bundle as bundle_group + from .commands.denied_dirs import denied_dirs as denied_dirs_group + from .commands.init import check_first_run, init_cmd, prompt_first_run_init + from .commands.module import module as module_group + from .commands.notify import notify as notify_group + from .commands.provider import provider as provider_group + from .commands.reset import reset as reset_cmd + from .commands.routing import routing_group + from .commands.run import register_run_command + from .commands.session import register_session_commands + from .commands.source import source as source_group + from .commands.tool import tool as tool_group + from .commands.update import update as update_cmd + from .commands.version import version as version_cmd + from .console import Markdown, console + from .dedicated_tty_input import close_dedicated_tty_input, get_dedicated_tty_input + from .effective_config import get_effective_config_summary + from .key_manager import KeyManager + from .lib.settings import AppSettings + from .provider_diagnostics import DEFAULT_TIMEOUT_S as _PROVIDER_DIAGNOSTIC_TIMEOUT_S + from .provider_diagnostics import format_model_line + from .provider_diagnostics import invoke_list_models + from .provider_diagnostics import test_provider_connectivity + from .session_runner import ( + SessionConfig, + create_initialized_session, + repair_transcript_if_needed, + ) + from .session_store import SessionStore + from .stdout_offload import patch_stdout_offloaded as patch_stdout + from .ui.dashboard_renderer import DashboardRenderer + from .ui.dashboard_renderer import _redact_value as _dr_redact_value + from .ui.error_display import display_llm_error, display_validation_error + from .ui.item_renderer import ItemRenderer + from .ui.log_filter import LLMErrorLogFilter + from .ui.completion import ( + Candidate, + SlashCompleter, + build_completion_snapshot, + longest_common_prefix, + ) + from .ui.view_policy import resolve_view + from .utils.error_format import escape_markup + from .utils.shell_completion import ( + can_safely_modify as _completion_can_safely_modify, + detect_shell as _completion_detect_shell, + get_shell_config_file as _completion_config_file, + install_completion_to_config as _install_completion, + is_click_completion_instruction, + is_completion_installed as _completion_is_installed, + manual_installation_instruction, + ) + from .utils.version import get_core_version, get_version + + logger = logging.getLogger(__name__) + # Attach this filter to console handlers, not the root logger: + # child provider/session log records would otherwise be duplicated. + _llm_error_filter = LLMErrorLogFilter() + # Completion never loads keys; ordinary startup retains that step. + if not os.environ.get("_AMPLIFIER_COMPLETE"): + KeyManager() + + # Apply the original decorators in their original order. Keep the + # group local until registration succeeds; never publish a callback. + group = click.pass_context(_cli_callback) + group = click.option( + "--install-completion", + is_flag=True, + flag_value="auto", + default=None, + help="Install completion for the shell detected from $SHELL (bash, zsh, or fish)", + )(group) + group = click.version_option( + version=f"{get_version()} (core {get_core_version()})", + prog_name="amplifier", + )(group) + group = click.group(name="cli", cls=AmplifierGroup, invoke_without_command=True)(group) + + from .commands.doctor import doctor + + group.add_command(doctor) + group.add_command(agents_group) + group.add_command(allowed_dirs_group) + group.add_command(denied_dirs_group) + group.add_command(bundle_group) + group.add_command(init_cmd) + group.add_command(module_group) + group.add_command(notify_group) + group.add_command(provider_group) + group.add_command(routing_group) + group.add_command(source_group) + group.add_command(tool_group) + group.add_command(update_cmd) + group.add_command(version_cmd) + group.add_command(reset_cmd) + + run_command = register_run_command( + group, + interactive_chat=interactive_chat, + execute_single=execute_single, + get_module_search_paths=get_module_search_paths, + check_first_run=check_first_run, + prompt_first_run_init=prompt_first_run_init, + ) + register_session_commands( + group, + interactive_chat=interactive_chat, + execute_single=execute_single, + get_module_search_paths=get_module_search_paths, + ) + _run_command = run_command + cli = group + _cli_ready = True + finally: + _cli_initializing = False + + +def __getattr__(name): + if name == "cli": + _initialize_cli() + return cli + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") def _guard_shared_venv_home() -> None: @@ -5098,6 +5139,12 @@ def _activate_home_env() -> None: def main(): """Main entry point.""" + if len(sys.argv) > 1 and sys.argv[1] == "doctor": + sys.dont_write_bytecode = True + from .commands.doctor import main as doctor_main + + return doctor_main() + _initialize_cli() completion_instruction = os.environ.get("_AMPLIFIER_COMPLETE") if completion_instruction: if not is_click_completion_instruction(completion_instruction): @@ -5113,5 +5160,11 @@ def main(): cli() +# Ordinary imports retain eager startup. Only the exact passive route defers +# initialization; main() always rechecks current argv rather than this decision. +if not (len(sys.argv) > 1 and sys.argv[1] == "doctor"): + _initialize_cli() + + if __name__ == "__main__": main() diff --git a/behaviors/cli-expertise.yaml b/behaviors/cli-expertise.yaml index 47ab1d39..94ea9971 100644 --- a/behaviors/cli-expertise.yaml +++ b/behaviors/cli-expertise.yaml @@ -9,6 +9,7 @@ bundle: agents: include: - app-cli:cli-expert + - app-cli:cli-doctor # Thin awareness pointer ONLY. # diff --git a/pyproject.toml b/pyproject.toml index 884de35b..45df9385 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,7 @@ dependencies = [ ] [project.scripts] -amplifier = "amplifier_app_cli.main:main" +amplifier = "amplifier_app_cli.__main__:main" [tool.uv] package = true diff --git a/tests/check_doctor_wheel.py b/tests/check_doctor_wheel.py new file mode 100644 index 00000000..1314b71c --- /dev/null +++ b/tests/check_doctor_wheel.py @@ -0,0 +1,211 @@ +"""Installed-wheel acceptance outside the checkout, without PYTHONPATH. + +Run only in a disposable wheel virtualenv. A synthetic wheel asks the actual +installer to generate the legacy amplifier console script, as another +distribution can do. No handwritten launcher replaces that acceptance check. +""" + +import base64 +import csv +import hashlib +import importlib.metadata as metadata +import io +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import sysconfig +import tempfile +import zipfile + + +FIXTURE_NAME = "doctor-legacy-launcher-fixture" + + +def _legacy_wheel(root): + info = "doctor_legacy_launcher_fixture-0.0.0.dist-info" + files = { + info + "/METADATA": ( + "Metadata-Version: 2.1\nName: " + FIXTURE_NAME + "\nVersion: 0.0.0\n\n" + ).encode(), + info + "/WHEEL": b"Wheel-Version: 1.0\nGenerator: synthetic-doctor-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n", + info + "/entry_points.txt": b"[console_scripts]\namplifier = amplifier_app_cli.main:main\n", + } + record = io.StringIO(newline="") + writer = csv.writer(record) + for name, data in files.items(): + digest = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).rstrip(b"=").decode() + writer.writerow((name, "sha256=" + digest, str(len(data)))) + writer.writerow((info + "/RECORD", "", "")) + files[info + "/RECORD"] = record.getvalue().encode() + wheel = root / "doctor_legacy_launcher_fixture-0.0.0-py3-none-any.whl" + with zipfile.ZipFile(wheel, "w") as archive: + for name, data in files.items(): + archive.writestr(name, data) + return wheel + + +def _run(command, args, env, project, code, *, collection=None): + result = subprocess.run( + command + ["doctor", *args], env=env, cwd=project, + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == code, result.stdout + result.stderr + errors = result.stderr.splitlines() + audit_lines = [line for line in errors if line.startswith("DOCTOR_AUDIT:")] + assert len(audit_lines) == 1, result.stderr + audit = json.loads(audit_lines[0].removeprefix("DOCTOR_AUDIT:")) + assert audit["violations"] == [], audit + assert audit["bytecode_suppressed"] and not audit["heavy_loaded"], audit + assert "synthetic-secret" not in result.stdout + result.stderr + assert "synthetic-private-value" not in result.stdout + result.stderr + errors.remove(audit_lines[0]) + if code == 2: + assert result.stdout == "" + assert errors == [ + "Usage: amplifier doctor [--format text|json] [--session SESSION_ID]", + "Error: invalid_doctor_arguments", + ] + else: + assert errors == [], errors + if collection is not None: + assert result.stdout.count("\n") == 1 + report = json.loads(result.stdout) + assert report["collection_status"] == collection + assert report["health"] == {"status": "not_assessed", "reason": "collection_only"} + return report + + +def _check_entry(command, root, env, ownership_pth): + project = root / "project" + app_home = Path(env["AMPLIFIER_HOME"]) + for args, code, status in ( + (["--format", "json"], 0, "complete"), ([], 0, None), + (["--help"], 0, None), + (["--session", "synthetic_session-01", "--format", "json"], 0, "complete"), + (["--format", "synthetic-secret"], 2, None), + (["--unknown=synthetic-secret"], 2, None), + (["--session", "../synthetic-secret"], 2, None), + (["--session"], 2, None), + (["synthetic-secret"], 2, None), + ): + _run(command, args, env, project, code, collection=status) + assert not app_home.exists() + assert not list(project.iterdir()) + + app_home.mkdir() + settings = app_home / "settings.yaml" + settings.write_text("bundle: [synthetic-secret", encoding="utf-8") + before = settings.read_bytes() + try: + _run(command, ["--format", "json"], env, project, 1, collection="partial") + assert settings.read_bytes() == before + assert list(app_home.iterdir()) == [settings] + finally: + settings.unlink() + app_home.rmdir() + + foreign = root / "synthetic-secret" / "cache" / "synthetic-0123456789abcdef" + ownership_pth.write_text(str(foreign) + "\n", encoding="utf-8") + try: + for override, status, code, collection in ( + (False, "refused", 1, "refused"), (True, "override", 0, "complete"), + ): + guarded_env = dict(env) + if override: + guarded_env["AMPLIFIER_ALLOW_SHARED_VENV"] = "1" + report = _run( + command, ["--format", "json"], guarded_env, project, code, + collection=collection, + ) + assert report["ownership"]["status"] == status + assert report["ownership"]["foreign_count"] == 1 + assert not app_home.exists() + assert not list(project.iterdir()) + finally: + ownership_pth.unlink() + + +def main(): + assert "PYTHONPATH" not in os.environ + dist = metadata.distribution("amplifier-app-cli") + location = Path(dist.locate_file("amplifier_app_cli")).resolve() + assert location.is_relative_to(Path(sys.prefix).resolve()) + direct = json.loads(dist.read_text("direct_url.json") or "{}") + assert not direct.get("dir_info", {}).get("editable") + entry = next(e for e in dist.entry_points if e.group == "console_scripts" and e.name == "amplifier") + assert entry.value == "amplifier_app_cli.__main__:main" + for relative in ( + "__main__.py", "main.py", "lib/diagnostics.py", "commands/doctor.py", + "_bundle/bundle.md", "_bundle/behaviors/cli-expertise.yaml", + "_bundle/agents/cli-doctor.md", + ): + assert (location / relative).is_file(), relative + + installer = shutil.which("uv") + assert installer is not None, "uv is required to generate the fixture launcher" + sitecustomize = Path(sysconfig.get_paths()["purelib"]) / "sitecustomize.py" + ownership_pth = sitecustomize.with_name("doctor-ownership-test.pth") + assert not sitecustomize.exists(), "only use a fresh dedicated wheel virtualenv" + assert not ownership_pth.exists() + probe = Path(__file__).with_name("doctor_safety_probe.py").read_bytes() + console = Path(sys.executable).parent / ("amplifier.exe" if os.name == "nt" else "amplifier") + original_console, original_mode = console.read_bytes(), console.stat().st_mode + fixture_installed = False + try: + with tempfile.TemporaryDirectory() as scratch: + root = Path(scratch) + (root / "project").mkdir() + (root / "home").mkdir() + env = { + key: value for key, value in os.environ.items() + if not key.startswith(("AMPLIFIER_", "_AMPLIFIER_")) + and key not in ("PYTHONPATH", "PYTHONDONTWRITEBYTECODE") + } + env.update( + HOME=str(root / "home"), USERPROFILE=str(root / "home"), + AMPLIFIER_HOME=str(root / "home" / "absent-amplifier"), + AMPLIFIER_HOME_ENV="1", AMPLIFIER_GIT_HOST="synthetic-private-value", + AMPLIFIER_MODULE_SYNTHETIC="synthetic-private-value", + ) + sitecustomize.write_bytes(probe) + for command in ( + [str(console)], [sys.executable, "-m", "amplifier_app_cli"], + [sys.executable, "-m", "amplifier_app_cli.main"], + ): + _check_entry(command, root, env, ownership_pth) + sitecustomize.unlink() + + # Install a valid fixture wheel, not a script crafted by this test. + wheel = _legacy_wheel(root) + subprocess.run( + [installer, "pip", "install", "--python", sys.executable, + "--no-deps", "--no-index", str(wheel)], + cwd=root, env=env, check=True, capture_output=True, text=True, + ) + fixture_installed = True + fixture = metadata.distribution(FIXTURE_NAME) + legacy = next(e for e in fixture.entry_points if e.name == "amplifier") + assert legacy.value == "amplifier_app_cli.main:main" + assert console.read_bytes() != original_console + sitecustomize.write_bytes(probe) + _check_entry([str(console)], root, env, ownership_pth) + finally: + ownership_pth.unlink(missing_ok=True) + sitecustomize.unlink(missing_ok=True) + try: + if fixture_installed: + subprocess.run( + [installer, "pip", "uninstall", "--python", sys.executable, FIXTURE_NAME], + check=True, capture_output=True, text=True, + ) + finally: + console.write_bytes(original_console) + console.chmod(original_mode) + print("Installed wheel: new console, module, legacy module, installer-generated legacy console, and doctor assets verified") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/doctor_safety_probe.py b/tests/doctor_safety_probe.py new file mode 100644 index 00000000..44c83b5b --- /dev/null +++ b/tests/doctor_safety_probe.py @@ -0,0 +1,125 @@ +"""Fresh-process instrumentation; also usable in an isolated wheel environment. + +Install this file as sitecustomize.py before starting the published entry. +The test harness, not the doctor, owns that setup. All attempted forbidden +operations are recorded even when application code swallows their exceptions. +""" + +import atexit +import json +import linecache # noqa: F401 -- Python 3.13 registers -c source before executing its body. +import os +import runpy # noqa: F401 -- Finish -m bootstrap before measuring app-owned activity. +import sys + +violations = [] +initial_bytecode = [] +owned_started = False +initial_fd_pending = False + + +def deny(code): + violations.append(code) + raise RuntimeError("doctor_probe_denied") + + +def forbidden_import(name): + return ( + name.startswith(("amplifier_foundation", "amplifier_core", "amplifier_module_")) + or (name.startswith("amplifier_app_cli") and name not in ( + "amplifier_app_cli", "amplifier_app_cli.__main__", "amplifier_app_cli.main", + "amplifier_app_cli.commands", "amplifier_app_cli.commands.doctor", + "amplifier_app_cli.lib", "amplifier_app_cli.lib.diagnostics", + "amplifier_app_cli.lib.venv_home_guard", "amplifier_app_cli.project_utils", + )) + ) + + +class BoundaryFinder: + def find_spec(self, fullname, path=None, target=None): + if forbidden_import(fullname): + deny("forbidden_import") + + +def audit(event, args): + global owned_started, initial_fd_pending + if event == "exec" and str(args[0].co_filename).replace("\\", "/").endswith("/amplifier_app_cli/__init__.py"): + owned_started = True + if event == "import": + if forbidden_import(args[0]): + deny("forbidden_import") + if event.startswith(("socket.", "subprocess.")) or event in ( + "os.system", "os.posix_spawn", "os.fork", "pty.spawn", + ): + deny("network_or_subprocess") + if event == "open": + path, mode, flags = args + path = str(path).replace("\\", "/") + writing = (isinstance(mode, str) and any(c in mode for c in "wax+")) or ( + isinstance(flags, int) and flags & (os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC | os.O_APPEND) + ) + if writing: + if not owned_started and "/amplifier_app_cli/__pycache__/__init__." in path: + initial_bytecode.append("initial_package_bytecode") + initial_fd_pending = True + elif not owned_started and isinstance(args[0], int) and initial_fd_pending: + initial_bytecode.append("initial_package_bytecode") + initial_fd_pending = False + else: + deny("write") + elif any(piece in path.split("/") for piece in ( + ".env", ".netrc", "keys.env", "keys.yaml", "keys.json", "credentials", "credentials.json", + "transcript.jsonl", "metadata.json", + )): + deny("credential_or_session_read") + if event in ( + "os.mkdir", "os.remove", "os.rename", "os.rmdir", "os.link", "os.symlink", + "os.chmod", "os.chown", "os.truncate", "tempfile.mkstemp", "tempfile.mkdtemp", + ): + path = str(args[0]).replace("\\", "/") if args else "" + if ( + event in ("os.mkdir", "os.rename") + and "/amplifier_app_cli/__pycache__" in path + and not owned_started + ): + initial_bytecode.append("initial_package_bytecode") + else: + deny("filesystem_mutation") + + +original_getitem = os._Environ.__getitem__ + + +def environ_getitem(self, key): + if any(word in str(key).upper() for word in ("API_KEY", "PASSWORD", "TOKEN", "CREDENTIAL")): + deny("credential_environment_read") + if str(key).startswith("AMPLIFIER_MODULE_") or key in ("AMPLIFIER_GIT_HOST", "AMPLIFIER_HOME_ENV"): + deny("environment_channel_value_read") + return original_getitem(self, key) + + +def environ_contains(self, key): + # Presence checks are allowed; unlike __getitem__, they return no value. + try: + original_getitem(self, key) + return True + except KeyError: + return False + + +os._Environ.__getitem__ = environ_getitem +os._Environ.__contains__ = environ_contains +sys.meta_path.insert(0, BoundaryFinder()) +sys.addaudithook(audit) + + +def finish(): + sys.stderr.write("DOCTOR_AUDIT:" + json.dumps({ + "violations": violations, + "initial_bytecode": initial_bytecode, + "bytecode_suppressed": sys.dont_write_bytecode, + "heavy_loaded": any(n.startswith(("amplifier_foundation", "amplifier_core")) for n in sys.modules), + }) + "\n") + + +atexit.register(finish) \ No newline at end of file diff --git a/tests/test_amplifier_home_isolation.py b/tests/test_amplifier_home_isolation.py index ed7aa4c0..b3d0bb79 100644 --- a/tests/test_amplifier_home_isolation.py +++ b/tests/test_amplifier_home_isolation.py @@ -151,11 +151,27 @@ def _violations(path: Path) -> list[str]: tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) aliases = _home_aliases(tree) found: list[str] = [] + # The passive doctor cannot import Foundation's eager package initializer. + # Permit only its one fallback expression, within current_target; behavioral + # parity against the installed Foundation helper lives in test_doctor.py. + passive_default_lines = set() + if path == _PACKAGE_ROOT / "lib" / "diagnostics.py": + target_function = next( + node for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "current_target" + ) + passive_default_lines = { + node.lineno for node in ast.walk(target_function) + if _joins_amplifier(node) and _is_path_home_call(node.left) + } + assert len(passive_default_lines) == 1 for node in ast.walk(tree): # Path.home() / ".amplifier" -- and the one-hop `home = Path.home()` # form, which reads innocently and is exactly how routing.py hid one. if _joins_amplifier(node): + if node.lineno in passive_default_lines: + continue left = node.left # type: ignore[attr-defined] if _is_path_home_call(left) or ( isinstance(left, ast.Name) and left.id in aliases diff --git a/tests/test_cli_doctor_agent.py b/tests/test_cli_doctor_agent.py new file mode 100644 index 00000000..4dfaaca7 --- /dev/null +++ b/tests/test_cli_doctor_agent.py @@ -0,0 +1,62 @@ +"""CLI-only doctor registration and on-demand composition, without preparation.""" + +from pathlib import Path + +import pytest +import yaml +from amplifier_foundation import load_bundle + + +ROOT = Path(__file__).resolve().parents[1] + + +def _frontmatter(path): + text = path.read_text(encoding="utf-8") + header, body = text.split("---", 2)[1:] + return yaml.safe_load(header), body + + +def test_cli_doctor_routing_and_existing_skill_boundary(): + metadata, body = _frontmatter(ROOT / "agents" / "cli-doctor.md") + description = metadata["meta"]["description"] + assert metadata["meta"]["name"] == "cli-doctor" + assert metadata["model_role"] == "general" + assert len(description) <= 400 + assert "USE WHEN" in description and "DO NOT USE" in description + assert "app-cli:cli-expert" in description + assert "amplifier-config" in body and "`load_skill`" in body + assert "child session" in body and "otherwise mark it unknown" in body + assert "Exit 0 means collection completed, not healthy" in body + assert "no repair verdict from passive collection" in body + assert "not that nothing needs fixing" in body + assert "does not authorize repair" in body + assert "Success criteria" in body + assert "amplifier doctor --format json" in body + skill = ROOT / "amplifier_app_cli" / "data" / "skills" / "amplifier-config" / "SKILL.md" + assert skill.is_file() + assert not (ROOT / "amplifier_app_cli" / "data" / "skills" / "cli-doctor").exists() + + +@pytest.mark.asyncio +async def test_cli_behavior_adds_agent_without_replacing_root(tmp_path): + # Local-only parsing/composition: do not prepare/install/mount anything. + root_file = tmp_path / "bundle.md" + root_file.write_text( + "---\nbundle:\n name: synthetic-root\n version: 0.1.0\n" + "providers:\n - module: provider-synthetic\n" + " config:\n priority: 7\n" + "session:\n synthetic: retained\n" + "---\nROOT-INSTRUCTION-MARKER\n", + encoding="utf-8", + ) + root = await load_bundle(root_file.as_uri()) + behavior = await load_bundle((ROOT / "behaviors" / "cli-expertise.yaml").as_uri()) + composed = root.compose(behavior) + assert composed.instruction == root.instruction + assert composed.providers == root.providers + assert composed.session == root.session + assert "app-cli:cli-doctor" in composed.agents + assert "Amplifier CLI Doctor" not in (composed.instruction or "") + parsed = yaml.safe_load((ROOT / "behaviors" / "cli-expertise.yaml").read_text()) + assert parsed["agents"]["include"] == ["app-cli:cli-expert", "app-cli:cli-doctor"] + assert parsed["context"]["include"] == ["app-cli:context/cli-awareness.md"] \ No newline at end of file diff --git a/tests/test_doctor.py b/tests/test_doctor.py new file mode 100644 index 00000000..e1119676 --- /dev/null +++ b/tests/test_doctor.py @@ -0,0 +1,359 @@ +"""Synthetic collector fixtures: nothing here comes from a real user home.""" + +import importlib.metadata as metadata +import json +import os +from pathlib import Path +from types import SimpleNamespace + +import pytest +import yaml +from click.testing import CliRunner + +from amplifier_app_cli.commands.doctor import doctor +from amplifier_app_cli.lib import diagnostics as diag +from amplifier_app_cli.lib import venv_home_guard as guard + +SECRET = "synthetic-private-sentinel-482" + + +@pytest.fixture +def target(tmp_path, monkeypatch): + project = tmp_path / "project" + project.mkdir() + home = tmp_path / "absent-home" + monkeypatch.chdir(project) + monkeypatch.setenv("AMPLIFIER_HOME", str(home)) + monkeypatch.delenv("AMPLIFIER_ALLOW_SHARED_VENV", raising=False) + for name in list(os.environ): + if name.startswith("AMPLIFIER_MODULE_"): + monkeypatch.delenv(name) + # Real guard, synthetic site-packages boundary. + site = tmp_path / "site-packages" + site.mkdir() + monkeypatch.setattr(guard, "site_packages_dir", lambda: site) + return diag.current_target() + + +def write_settings(target, text): + target.home.mkdir(exist_ok=True) + (target.home / "settings.yaml").write_text(text, encoding="utf-8") + + +def safe_report(target): + report = diag.collect_diagnostics(target) + assert SECRET not in json.dumps(report) + assert SECRET not in diag.render_text(report) + return report + + +def test_missing_files_complete_without_creating_home(target): + report = safe_report(target) + assert report["schema_version"] == 1 + assert report["mode"] == "passive" + assert report["collection_status"] == "complete" + assert report["health"] == {"status": "not_assessed", "reason": "collection_only"} + assert "Health: not assessed" in diag.render_text(report) + assert [row["status"] for row in report["scopes"]] == ["missing"] * 3 + ["not_selected"] + assert report["effective_settings"]["reason"] == "effective_resolution_not_performed" + assert report["live_session"]["reason"] == "not_observed" + assert not target.home.exists() + + +@pytest.mark.parametrize("text,status", [ + ("", "empty"), ("{}", "empty"), ("null", "empty"), + ("[]", "invalid_shape"), ("false", "invalid_shape"), + ("bundle: [" + SECRET, "invalid_yaml"), + (f"a: {SECRET}\na: 2", "unsupported_yaml"), + (f"a: &x [{SECRET}]\nb: *x", "unsupported_yaml"), + (f"a: !!str {SECRET}", "unsupported_yaml"), + (f"a: !synthetic {SECRET}", "unsupported_yaml"), + ("? [bad, key]\n: value", "invalid_yaml"), + ("a: " + "x" * 16385, "too_large"), + ("a: " + "[" * 34 + "0" + "]" * 34, "too_large"), + ("a: [" + ",".join("0" for _ in range(10001)) + "]", "too_large"), + ("#" * (diag.SETTINGS_LIMIT + 1), "too_large"), +]) +def test_bounded_settings(target, text, status): + write_settings(target, text) + report = safe_report(target) + assert report["scopes"][0]["status"] == status + assert report["collection_status"] == ("complete" if status == "empty" else "partial") + if status != "empty": + assert {"code": "settings_" + status, "scope": "global"} in report["issues"] + assert not any(row["scope"] == "global" for row in report["declarations"]) + if status == "unsupported_yaml": + assert {"code": "settings_invalid_yaml", "scope": "global"} not in report["issues"] + + +def test_invalid_utf8_is_safely_partial(target): + target.home.mkdir() + (target.home / "settings.yaml").write_bytes(b"\xff") + report = safe_report(target) + assert report["scopes"][0]["status"] == "invalid_yaml" + assert report["collection_status"] == "partial" + assert {"code": "settings_invalid_yaml", "scope": "global"} in report["issues"] + + +def test_all_scopes_and_exact_session_layout(target): + from amplifier_app_cli.lib.settings import SettingsPaths + + selected = diag.DiagnosticTarget(target.cwd, target.home, "session_01") + paths = SettingsPaths.with_session("session_01", diag.get_project_slug()) + expected = [ + paths.global_settings, paths.project_settings, paths.local_settings, paths.session_settings, + ] + for index, path in enumerate(expected): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"routing: {{matrix: synthetic-{index}}}", encoding="utf-8") + report = safe_report(selected) + assert [r["path"] for r in report["scopes"]] == [str(p) for p in expected] + assert [r["status"] for r in report["scopes"]] == ["valid"] * 4 + assert report["target"]["session_selected"] is True + + +@pytest.mark.parametrize("home_value", [None, "", "~/alternate", "./relative"]) +def test_home_parity_with_installed_foundation(target, monkeypatch, home_value): + from amplifier_foundation.paths.resolution import get_amplifier_home + + if home_value is None: + monkeypatch.delenv("AMPLIFIER_HOME", raising=False) + else: + monkeypatch.setenv("AMPLIFIER_HOME", home_value) + assert diag.current_target().home == get_amplifier_home().resolve() + + +@pytest.mark.parametrize("value", ["../escape", "/absolute", "x/y", "a.b", "", "-a", "x" * 201, "a\nb", "é", "x\\y"]) +def test_unsafe_session_rejected_without_echo(target, value): + assert not diag.valid_session_id(value) + with pytest.raises(ValueError, match="^invalid_session_id$"): + diag.current_target(value) + result = CliRunner().invoke(doctor, ["--session", value]) + assert result.exit_code == 2 + assert value not in result.output or not value # Click error uses fixed callback text. + + +def test_declarations_are_opaque_and_equality_shared(target, monkeypatch): + source = f"https://user:{SECRET}@example.test/repo?token={SECRET}#private" + settings = { + SECRET: {SECRET: SECRET}, + "bundle": {"active": SECRET, "app": [SECRET, source]}, + "routing": {"matrix": SECRET}, + "sources": {"modules": {SECRET: source}, "bundles": {SECRET: source}}, + "overrides": { + SECRET: {"source": source, "config": {"api_key": SECRET}}, + "modules": {SECRET: source}, "bundles": [SECRET], + }, + "modules": {"providers": [{"id": SECRET, "module": SECRET, "source": source}]}, + "config": {"providers": [{"module": SECRET, "source": source}]}, + } + write_settings(target, yaml.safe_dump(settings, sort_keys=False)) + monkeypatch.setenv("AMPLIFIER_MODULE_" + SECRET, source) + monkeypatch.setenv("AMPLIFIER_GIT_HOST", source) + monkeypatch.setenv("AMPLIFIER_HOME_ENV", source) + report = safe_report(target) + rows = report["declarations"] + active = next(r for r in rows if r["container"] == "bundle" and r["member"] == "active") + matrix = next(r for r in rows if r["member"] == "matrix") + provider = next(r for r in rows if r["container"] == "modules.providers") + assert active["value_ref"] == matrix["value_ref"] == provider["id_ref"] == provider["module_ref"] + sources = [r for r in rows if r["member"] == "source"] + assert len({r["value_ref"] for r in sources}) == 1 + assert all("index" in r for r in sources) + assert provider["source_kind"] == "http" + assert next(r for r in rows if r["container"] == "overrides" and r["member"] == "modules")["shape"] == "mapping" + assert report["environment_channels"]["module_count"] == 1 + assert report["environment_channels"]["AMPLIFIER_HOME_ENV"] == {"present": True} + assert report["collection_status"] == "complete" + + +@pytest.mark.parametrize("value,shape", [ + (None, "null"), (False, "false"), (True, "true"), ("", "empty_string"), + ([], "empty_list"), ({}, "empty_mapping"), (12, "integer"), (2.5, "number"), +]) +def test_declaration_shapes_preserved(target, value, shape): + write_settings(target, yaml.safe_dump({"bundle": {"active": value}})) + report = safe_report(target) + assert next(r for r in report["declarations"] if r["member"] == "active")["shape"] == shape + + +def test_container_null_distinguished_from_absent(target): + write_settings(target, "bundle: null") + rows = safe_report(target)["declarations"] + assert next(r for r in rows if r["member"] == "active")["container_shape"] == "null" + assert next(r for r in rows if r["member"] == "matrix")["container_shape"] == "absent" + + +def test_limits_are_explicit(target, monkeypatch): + write_settings(target, yaml.safe_dump({"sources": {"modules": {f"module-{i}": SECRET for i in range(210)}}})) + for i in range(210): + monkeypatch.setenv(f"AMPLIFIER_MODULE_SYNTHETIC_{i}", SECRET) + report = safe_report(target) + assert len(report["declarations"]) == 200 + assert len(report["environment_channels"]["modules"]) == 200 + assert report["environment_channels"]["module_count"] == 210 + assert report["collection_status"] == "partial" + assert {"declarations_truncated", "environment_channels_truncated"} <= {r["code"] for r in report["issues"]} + + +def test_regular_files_only(target): + target.home.mkdir() + (target.home / "settings.yaml").mkdir() + assert safe_report(target)["scopes"][0]["status"] == "unsupported_file" + + +@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="POSIX FIFO") +def test_fifo_is_not_opened(target): + target.home.mkdir() + os.mkfifo(target.home / "settings.yaml") + assert safe_report(target)["scopes"][0]["status"] == "unsupported_file" + + +def test_leaf_and_session_parent_symlinks_rejected(target, tmp_path): + real = tmp_path / "real.yaml" + real.write_text("bundle: {active: synthetic}", encoding="utf-8") + target.home.mkdir() + try: + (target.home / "settings.yaml").symlink_to(real) + except OSError: + pytest.skip("symlink creation not permitted") + assert safe_report(target)["scopes"][0]["status"] == "unsupported_symlink" + (target.home / "projects").symlink_to(tmp_path, target_is_directory=True) + selected = diag.DiagnosticTarget(target.cwd, target.home, "synthetic") + assert safe_report(selected)["scopes"][3]["status"] == "unsupported_symlink" + + +def test_unreadable_and_errors_do_not_leak(target, monkeypatch): + write_settings(target, "bundle: {active: synthetic}") + original = os.open + + def fail(path, flags, *args, **kwargs): + if Path(path) == target.home / "settings.yaml": + raise PermissionError(SECRET) + return original(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", fail) + assert safe_report(target)["scopes"][0]["status"] == "unreadable" + + +def test_guard_refusal_stops_collection_and_override_is_reported(target, monkeypatch): + site = guard.site_packages_dir() + foreign = site.parent / SECRET / "cache" / "synthetic-0123456789abcdef" + (site / "synthetic.pth").write_text(str(foreign), encoding="utf-8") + report = safe_report(target) + assert report["collection_status"] == "refused" + assert report["ownership"]["status"] == "refused" + assert report["ownership"]["foreign_count"] == 1 + assert report["runtime"] is None + assert report["declarations"] == [] + monkeypatch.setenv("AMPLIFIER_ALLOW_SHARED_VENV", "1") + report = safe_report(target) + assert report["collection_status"] == "complete" + assert report["ownership"]["status"] == "override" + assert not target.home.exists() + + +def test_unknown_guard_fails_closed(target, monkeypatch): + def fail(*args, **kwargs): + raise RuntimeError(SECRET) + + monkeypatch.setattr(guard, "detect_home_conflict", fail) + report = safe_report(target) + assert report["collection_status"] == "refused" + assert report["ownership"]["status"] == "unknown" + assert report["runtime"] is None + assert report["issues"] == [{"code": "ownership_unknown"}] + + +@pytest.mark.parametrize("suffix", ["\x1b[31m", "\u202e", "/https://user:password@example.test", "?token=synthetic"]) +def test_unsafe_path_withheld(target, tmp_path, suffix): + path = tmp_path / ("unsafe" + suffix) + report = safe_report(diag.DiagnosticTarget(path, target.home)) + assert report["target"]["cwd"] is None + assert {"code": "path_withheld"} in report["issues"] + assert suffix not in diag.render_text(report) + assert report["collection_status"] == "partial" + + +def fake_distribution(tmp_path, monkeypatch, direct_url, version="1.2.3"): + root = tmp_path / "installed" + info = root / "synthetic.dist-info" + info.mkdir(parents=True) + path = info / "direct_url.json" + path.write_text(json.dumps(direct_url), encoding="utf-8") + dist = SimpleNamespace( + version=version, + files=[metadata.PackagePath("synthetic.dist-info/direct_url.json")], + locate_file=lambda entry: root / entry, + ) + monkeypatch.setattr(metadata, "distribution", lambda name: dist) + return path + + +def test_distribution_metadata_allowlist(target, tmp_path, monkeypatch): + fake_distribution(tmp_path, monkeypatch, { + "url": f"git+https://user:{SECRET}@example.test/repo?token={SECRET}", + "vcs_info": {"vcs": SECRET, "requested_revision": SECRET, "commit_id": "abcdef0123456789"}, + "dir_info": {"editable": True}, SECRET: SECRET, + }) + report = safe_report(target) + direct = report["runtime"]["distributions"][0]["direct_url"] + assert direct["commit"] == "abcdef0123456789" + assert direct["local_target"] is None + assert direct["editable"] is True + + +@pytest.mark.parametrize("url,expected", [ + ("file:///synthetic/a%20b", "/synthetic/a b"), + ("file:///synthetic/%2541", None), # no second decoding or percent-shaped disclosure + ("file://authority/synthetic", None), + ("file:///synthetic?token=hidden", None), + ("file:///synthetic#fragment", None), +]) +def test_direct_url_local_paths_only(target, tmp_path, monkeypatch, url, expected): + fake_distribution(tmp_path, monkeypatch, {"url": url}) + report = safe_report(target) + actual = report["runtime"]["distributions"][0]["direct_url"]["local_target"] + if os.name == "nt" and expected is not None: + # A POSIX-root URL has no Windows drive and is not native absolute. + expected = None + assert actual == expected + + +def test_direct_url_native_absolute_target(target, tmp_path, monkeypatch): + fake_distribution(tmp_path, monkeypatch, {"url": target.cwd.as_uri(), "dir_info": {"editable": True}}) + report = safe_report(target) + direct = report["runtime"]["distributions"][0]["direct_url"] + assert direct["local_target"] == str(target.cwd) + assert direct["editable"] is True + + +@pytest.mark.parametrize("kind", ["large", "invalid_json", "version", "local_version", "error"]) +def test_bad_metadata_is_partial_without_payload(target, tmp_path, monkeypatch, kind): + version = SECRET if kind == "version" else "1.0+" + SECRET if kind == "local_version" else "1.0" + path = fake_distribution(tmp_path, monkeypatch, {"url": SECRET}, version=version) + if kind == "large": + path.write_text(SECRET * 4000, encoding="utf-8") + if kind == "invalid_json": + path.write_text(SECRET, encoding="utf-8") + if kind == "error": + def fail(name): + raise RuntimeError(SECRET) + monkeypatch.setattr(metadata, "distribution", fail) + report = safe_report(target) + assert report["collection_status"] == "partial" + + +def test_text_and_json_exit_contract(target): + runner = CliRunner() + result = runner.invoke(doctor, ["--format", "json"]) + assert result.exit_code == 0 + assert result.output.count("\n") == 1 + assert json.loads(result.output)["collection_status"] == "complete" + write_settings(target, "bundle: [") + result = runner.invoke(doctor, ["--format", "json"]) + assert result.exit_code == 1 + assert json.loads(result.output)["collection_status"] == "partial" + result = runner.invoke(doctor) + assert result.exit_code == 1 + assert "not a health verdict" in result.output \ No newline at end of file diff --git a/tests/test_doctor_entrypoint.py b/tests/test_doctor_entrypoint.py new file mode 100644 index 00000000..e99acf20 --- /dev/null +++ b/tests/test_doctor_entrypoint.py @@ -0,0 +1,514 @@ +"""Exercise installed public entries in fresh interpreters, not main mocks.""" + +import ast +import importlib.metadata as metadata +import json +import os +from pathlib import Path +import subprocess +import sys + +import pytest + +PROBE = Path(__file__).with_name("doctor_safety_probe.py") + + +def test_bytecode_switch_precedes_nonbootstrap_package_imports(): + # A preloaded audit helper can mask stdlib bytecode attempts. Keep the + # package-owned switch before imports other than the bootstrap sys module. + package = Path(__file__).resolve().parents[1] / "amplifier_app_cli" / "__init__.py" + nodes = ast.parse(package.read_text(encoding="utf-8")).body + assert isinstance(nodes[0], ast.Expr) # Module docstring. + assert isinstance(nodes[1], ast.Import) + assert [alias.name for alias in nodes[1].names] == ["sys"] + assert isinstance(nodes[2], ast.If) + assert ast.unparse(nodes[2].body[0]) == "sys.dont_write_bytecode = True" + + +def _entry(kind): + if kind == "module": + return [sys.executable, "-m", "amplifier_app_cli"] + if kind == "legacy_module": + return [sys.executable, "-m", "amplifier_app_cli.main"] + launcher = Path(sys.executable).parent / ("amplifier.exe" if os.name == "nt" else "amplifier") + assert launcher.exists(), "test requires the installed project console entry" + return [str(launcher)] + + +def _run(tmp_path, kind, args, *, settings=None): + project = tmp_path / "project" + project.mkdir() + home = tmp_path / "home" + home.mkdir() + app_home = home / "absent-amplifier" + if settings is not None: + app_home.mkdir() + (app_home / "settings.yaml").write_text(settings, encoding="utf-8") + instrumentation = tmp_path / "instrumentation" + instrumentation.mkdir() + (instrumentation / "sitecustomize.py").write_bytes(PROBE.read_bytes()) + # This contains only instrumentation, not a source checkout/dependency override. + env = { + key: value for key, value in os.environ.items() + if not key.startswith(("AMPLIFIER_", "_AMPLIFIER_")) + and key not in ("PYTHONPATH", "PYTHONDONTWRITEBYTECODE") + } + env.update({ + "HOME": str(home), "USERPROFILE": str(home), "AMPLIFIER_HOME": str(app_home), + "PYTHONPATH": str(instrumentation), + "SYNTHETIC_API_KEY": "synthetic-secret-never-read", + "AMPLIFIER_HOME_ENV": "1", "AMPLIFIER_GIT_HOST": "synthetic-private-value", + "AMPLIFIER_MODULE_SYNTHETIC": "synthetic-private-value", + }) + before = sorted(str(p.relative_to(project)) for p in project.rglob("*")) + result = subprocess.run(_entry(kind) + ["doctor", *args], env=env, cwd=project, text=True, capture_output=True, timeout=30) + assert sorted(str(p.relative_to(project)) for p in project.rglob("*")) == before + assert app_home.exists() == (settings is not None) + lines = result.stderr.splitlines() + audit = json.loads(next(line.removeprefix("DOCTOR_AUDIT:") for line in lines if line.startswith("DOCTOR_AUDIT:"))) + assert audit["violations"] == [], result.stderr + assert audit["bytecode_suppressed"] is True + assert audit["heavy_loaded"] is False + assert "synthetic-private-value" not in result.stdout + result.stderr + return result, [line for line in lines if not line.startswith("DOCTOR_AUDIT:")] + + +@pytest.mark.parametrize("kind", ["console", "module", "legacy_module"]) +@pytest.mark.parametrize("args,code", [ + (["--format", "json"], 0), + ([], 0), + (["--help"], 0), + (["--session", "synthetic_session-01", "--format", "json"], 0), + (["--format", "synthetic-secret"], 2), + (["--format", "\x1b[31msynthetic-secret"], 2), + (["--unknown=synthetic-secret"], 2), + (["--session", "../synthetic-secret"], 2), + (["--session"], 2), + (["synthetic-secret"], 2), +]) +def test_published_entries_are_passive(tmp_path, kind, args, code): + result, errors = _run(tmp_path, kind, args) + assert result.returncode == code, result.stdout + result.stderr + assert "synthetic-secret" not in result.stdout + result.stderr + if code == 2: + assert result.stdout == "" + assert errors == [ + "Usage: amplifier doctor [--format text|json] [--session SESSION_ID]", + "Error: invalid_doctor_arguments", + ] + else: + assert errors == [] + if args == ["--format", "json"]: + assert result.stdout.count("\n") == 1 + assert json.loads(result.stdout)["collection_status"] == "complete" + + +@pytest.mark.parametrize("kind", ["console", "module", "legacy_module"]) +def test_real_partial_entry_returns_one(tmp_path, kind): + result, errors = _run(tmp_path, kind, ["--format", "json"], settings="bundle: [synthetic-secret") + assert result.returncode == 1 + assert json.loads(result.stdout)["collection_status"] == "partial" + assert "synthetic-secret" not in result.stdout + result.stderr + assert errors == [] + + +def test_console_metadata_points_to_dispatcher(): + dist = metadata.distribution("amplifier-app-cli") + entry = next(e for e in dist.entry_points if e.group == "console_scripts" and e.name == "amplifier") + assert entry.value == "amplifier_app_cli.__main__:main" + + +def test_importing_legacy_main_on_doctor_route_does_not_dispatch(tmp_path): + instrumentation = tmp_path / "instrumentation" + instrumentation.mkdir() + (instrumentation / "sitecustomize.py").write_bytes(PROBE.read_bytes()) + env = dict(os.environ, PYTHONPATH=str(instrumentation)) + env.pop("PYTHONDONTWRITEBYTECODE", None) + code = """ +import sys +sys.argv = ['amplifier', 'doctor', '--format', 'json'] +module = __import__('amplifier_app_cli.main', fromlist=['main']) +assert module.__name__ == 'amplifier_app_cli.main' +assert module.__file__.endswith('main.py') +assert callable(module.main) +assert 'amplifier_app_cli.commands.doctor' not in sys.modules +assert 'amplifier_app_cli.lib.diagnostics' not in sys.modules +assert not hasattr(module, 'KeyManager') +assert not hasattr(module, '_llm_error_filter') +assert not hasattr(module, 'unknown_export') +import amplifier_app_cli as package +import amplifier_app_cli.commands as commands +from amplifier_app_cli import main +assert main is module.main +assert package.main is main +assert not module._cli_ready +assert not hasattr(package, 'unknown_export') +assert not hasattr(commands, 'unknown_export') +assert package.__all__ == ['cli', 'main', '__version__'] +assert 'register_run_command' in commands.__all__ +assert not module._cli_ready +""" + result = subprocess.run( + [sys.executable, "-c", code], cwd=tmp_path, env=env, + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stderr + assert result.stdout == "" + lines = result.stderr.splitlines() + assert len(lines) == 1 and lines[0].startswith("DOCTOR_AUDIT:") + audit = json.loads(lines[0].removeprefix("DOCTOR_AUDIT:")) + assert audit["violations"] == [] + assert audit["bytecode_suppressed"] and not audit["heavy_loaded"] + + +def test_probe_records_attempts_even_when_exceptions_are_swallowed(tmp_path): + code = """ +import importlib, os, runpy, socket, subprocess, sys, tempfile +tempfile.tempdir = os.getcwd() +runpy.run_path(sys.argv[1]) +attempts = ( + lambda: open('synthetic-write', 'w'), + lambda: os.mkdir('synthetic-directory'), + lambda: tempfile.mkstemp(), + lambda: socket.socket(), + lambda: subprocess.run([sys.executable, '-c', 'pass']), + lambda: importlib.import_module('amplifier_foundation'), + lambda: open('.env'), + lambda: open('keys.env'), + lambda: os.environ.get('SYNTHETIC_API_KEY'), + lambda: os.environ.get('AMPLIFIER_GIT_HOST'), +) +for attempt in attempts: + try: + attempt() + except Exception: + pass +""" + result = subprocess.run( + [sys.executable, "-c", code, str(PROBE)], cwd=tmp_path, + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stderr + line = next(line for line in result.stderr.splitlines() if line.startswith("DOCTOR_AUDIT:")) + audit = json.loads(line.removeprefix("DOCTOR_AUDIT:")) + assert set(audit["violations"]) == { + "write", "filesystem_mutation", "network_or_subprocess", "forbidden_import", + "credential_or_session_read", "credential_environment_read", + "environment_channel_value_read", + } + assert audit["violations"].count("credential_or_session_read") == 2 + assert not list(tmp_path.iterdir()) + + +def _run_import_check(tmp_path, code, *args): + home = tmp_path / "home" + home.mkdir() + env = { + name: value for name, value in os.environ.items() + if not name.startswith(("AMPLIFIER_", "_AMPLIFIER_")) and name != "PYTHONPATH" + } + env.update(HOME=str(home), USERPROFILE=str(home), AMPLIFIER_HOME=str(home / "app")) + result = subprocess.run( + [sys.executable, "-c", code, *args], cwd=tmp_path, env=env, + capture_output=True, text=True, timeout=60, + ) + assert result.returncode == 0, result.stdout + result.stderr + return result + + +@pytest.mark.parametrize("route", ["ordinary", "foreign_doctor"]) +@pytest.mark.parametrize("order", ["package_first", "main_first", "module_cli", "command_first", "submodule_first"]) +def test_ordinary_exports_and_command_identity(order, route, tmp_path): + code = """ +import importlib +import sys +order, route = sys.argv[1:] +sys.argv = ['foreign-program', 'doctor'] if route == 'foreign_doctor' else ['foreign-program'] +if order in ('main_first', 'module_cli'): + module = importlib.import_module('amplifier_app_cli.main') + if order == 'module_cli': + import click + assert isinstance(module.cli, click.Group) +elif order == 'command_first': + from amplifier_app_cli.commands import register_run_command +elif order == 'submodule_first': + for name in ('provider', 'module', 'tool', 'source'): + importlib.import_module('amplifier_app_cli.commands.' + name) +import amplifier_app_cli as package +from amplifier_app_cli import cli, main +module = importlib.import_module('amplifier_app_cli.main') +import click +import amplifier_app_cli.commands as commands +from amplifier_app_cli.commands import auto_init_from_env, register_run_command +from amplifier_app_cli.commands.doctor import doctor +assert callable(main) +assert package.main is main is module.main +assert package.cli is cli is module.cli +assert isinstance(cli, click.Group) and cli.name == 'cli' +assert module.cli.commands['doctor'] is doctor +assert callable(auto_init_from_env) and callable(register_run_command) +assert package.__all__ == ['cli', 'main', '__version__'] +expected = { + 'auto_init_from_env': 'init', 'check_first_run': 'init', 'init_cmd': 'init', + 'prompt_first_run_init': 'init', 'module': 'module', 'provider': 'provider', + 'register_run_command': 'run', 'register_session_commands': 'session', + 'source': 'source', 'tool': 'tool', +} +assert set(commands.__all__) == set(expected) +for name, origin in expected.items(): + child = importlib.import_module('amplifier_app_cli.commands.' + origin) + assert getattr(commands, name) is getattr(child, name), name +assert 'amplifier_app_cli.truststore_shim' in sys.modules +params = tuple(cli.params) +registered = dict(cli.commands) +run = module._run_command +error_filter = module._llm_error_filter +def no_keys(): + raise AssertionError('repeat initialization loaded keys') +module.KeyManager = no_keys +for _ in range(3): + module._initialize_cli() + from amplifier_app_cli import cli as again, main as dispatch + assert again is cli and dispatch is main +assert tuple(cli.params) == params +assert len({param.name for param in params}) == len(params) == 2 +assert cli.commands == registered +assert module._run_command is run +assert module._llm_error_filter is error_filter +""" + _run_import_check(tmp_path, code, order, route) + + +@pytest.mark.parametrize("route", ["ordinary", "foreign_doctor"]) +def test_submodule_first_dispatcher_only_preserves_package_identity(route, tmp_path): + code = """ +import importlib +import sys +from types import ModuleType +route = sys.argv[1] +sys.argv = ['foreign-program', 'doctor'] if route == 'foreign_doctor' else ['foreign-program'] +child = importlib.import_module('amplifier_app_cli.main') +package = sys.modules['amplifier_app_cli'] +spec, path, exports = package.__spec__, package.__path__, package.__all__ +if route == 'foreign_doctor': + assert package.__dict__['main'] is child +from amplifier_app_cli import main +assert main is child.main +for _ in range(3): + assert package.main is main +assert isinstance(package, ModuleType) +assert importlib.import_module('amplifier_app_cli') is package +assert sys.modules[package.__name__] is package +assert sys.modules[child.__name__] is child +assert package.__spec__ is spec and spec.name == 'amplifier_app_cli' +assert package.__path__ is path and spec.submodule_search_locations is path +assert package.__all__ is exports +assert exports == ['cli', 'main', '__version__'] +sentinel = object() +package.main = sentinel +assert package.main is sentinel +package.main = child +assert package.main is main +if route == 'foreign_doctor': + assert not child._cli_ready and not hasattr(child, 'KeyManager') + assert 'amplifier_foundation' not in sys.modules + assert 'amplifier_app_cli.truststore_shim' not in sys.modules +""" + _run_import_check(tmp_path, code, route) + + +@pytest.mark.parametrize("route", ["ordinary", "foreign_doctor"]) +@pytest.mark.parametrize("name", ["module", "provider", "source", "tool"]) +def test_submodule_first_same_name_command_only(name, route, tmp_path): + code = """ +import importlib +import sys +from types import ModuleType +name, route = sys.argv[1:] +sys.argv = ['foreign-program', 'doctor'] if route == 'foreign_doctor' else ['foreign-program'] +child = importlib.import_module('amplifier_app_cli.commands.' + name) +package = sys.modules['amplifier_app_cli.commands'] +spec, path, exports = package.__spec__, package.__path__, package.__all__ +if route == 'foreign_doctor': + assert package.__dict__[name] is child +if name == 'module': + from amplifier_app_cli.commands import module as exported +elif name == 'provider': + from amplifier_app_cli.commands import provider as exported +elif name == 'source': + from amplifier_app_cli.commands import source as exported +else: + from amplifier_app_cli.commands import tool as exported +import click +assert isinstance(exported, click.Group) +assert exported is getattr(child, name) +for _ in range(3): + package._load_exports() + assert getattr(package, name) is exported +assert isinstance(package, ModuleType) +assert importlib.import_module('amplifier_app_cli.commands') is package +assert sys.modules[package.__name__] is package +assert sys.modules[child.__name__] is child +assert package.__spec__ is spec and spec.name == 'amplifier_app_cli.commands' +assert package.__path__ is path and spec.submodule_search_locations is path +assert package.__all__ is exports and name in exports +sentinel = object() +setattr(package, name, sentinel) +package._load_exports() +assert getattr(package, name) is sentinel +# Even when exports are already loaded, a child collision returns its command. +setattr(package, name, child) +assert getattr(package, name) is exported +assert getattr(package, name) is exported +if route == 'foreign_doctor': + assert 'amplifier_app_cli.main' not in sys.modules +""" + _run_import_check(tmp_path, code, name, route) + + +def test_concurrent_submodule_collision_reads_finish(tmp_path): + code = """ +import importlib +import sys +import threading +import time +sys.argv = ['foreign-program', 'doctor'] +child = importlib.import_module('amplifier_app_cli.main') +command_child = importlib.import_module('amplifier_app_cli.commands.provider') +root = sys.modules['amplifier_app_cli'] +commands = sys.modules['amplifier_app_cli.commands'] +barrier = threading.Barrier(4) +errors = [] +results = [] +def read_exports(): + try: + barrier.wait(timeout=10) + for _ in range(3): + assert root.main is child.main + assert commands.provider is command_child.provider + results.append(True) + except BaseException as error: + errors.append(repr(error)) +threads = [threading.Thread(target=read_exports, daemon=True) for _ in range(4)] +for thread in threads: + thread.start() +deadline = time.monotonic() + 20 +for thread in threads: + thread.join(max(0, deadline - time.monotonic())) +assert not any(thread.is_alive() for thread in threads), 'export reads deadlocked' +assert errors == [], errors +assert len(results) == len(threads) +assert not child._cli_ready +""" + _run_import_check(tmp_path, code) + + +def test_package_dispatcher_alone_stays_passive(tmp_path): + code = """ +import sys +sys.argv = ['foreign-program', 'doctor'] +from amplifier_app_cli import main +module = sys.modules['amplifier_app_cli.main'] +assert main is module.main +assert not module._cli_ready +assert not hasattr(module, 'KeyManager') +assert not hasattr(module, 'unrecognized') +assert 'amplifier_foundation' not in sys.modules +assert 'amplifier_core' not in sys.modules +assert 'amplifier_app_cli.truststore_shim' not in sys.modules +assert 'amplifier_app_cli.commands.doctor' not in sys.modules +""" + _run_import_check(tmp_path, code) + + +@pytest.mark.parametrize("completion", ["ordinary", "completion"]) +def test_cached_passive_import_initializes_for_current_normal_argv(tmp_path, completion): + code = """ +import importlib +import os +import sys +mode = sys.argv[1] +sys.argv = ['foreign-program', 'doctor'] +module = importlib.import_module('amplifier_app_cli.main') +import amplifier_app_cli as package +assert not module._cli_ready +events = [] +key_calls = [] +original_tls = package._ensure_tls_shim +def tls(): + events.append('tls') + original_tls() +package._ensure_tls_shim = tls +class ImportOrder: + def find_spec(self, fullname, path=None, target=None): + if fullname.startswith(('amplifier_core', 'amplifier_foundation', 'prompt_toolkit')): + assert 'tls' in events, fullname +finder = ImportOrder() +sys.meta_path.insert(0, finder) +def profile(frame, event, arg): + if (event == 'call' and frame.f_code.co_name == '__init__' + and frame.f_globals.get('__name__') == 'amplifier_app_cli.key_manager'): + key_calls.append('keys') +sys.setprofile(profile) +original_initialize = module._initialize_cli +groups = [] +def initialize(): + original_initialize() + groups.append(module.cli) + events.append('initialized') + module.cli = lambda: events.append('cli') +module._initialize_cli = initialize +for name, label in ( + ('_ensure_utf8_output', 'utf8'), ('_configure_console_logging', 'logging'), + ('_attach_llm_error_filter', 'filter'), ('_guard_shared_venv_home', 'guard'), + ('_activate_home_env', 'activate'), +): + setattr(module, name, lambda label=label: events.append(label)) +if mode == 'completion': + os.environ['_AMPLIFIER_COMPLETE'] = 'bash_complete' +sys.argv = ['foreign-program', '--help'] +module.main() +sys.setprofile(None) +sys.meta_path.remove(finder) +assert module._cli_ready +assert key_calls == ([] if mode == 'completion' else ['keys']), key_calls +expected = ['initialized', 'cli'] if mode == 'completion' else [ + 'initialized', 'utf8', 'logging', 'filter', 'guard', 'activate', 'cli', +] +assert [event for event in events if event != 'tls'] == expected, events +import click +assert isinstance(groups[0], click.Group) +patched_cli = module.cli +original_initialize() +assert module.cli is patched_cli +assert module.interactive_chat.__globals__ is module.__dict__ +assert module.execute_single.__globals__ is module.__dict__ + +# Once loaded, a later doctor dispatch must not rerun ordinary initialization. +doctor_module = importlib.import_module('amplifier_app_cli.commands.doctor') +doctor_module.main = lambda: 'passive' +def forbidden(): + raise AssertionError('doctor initialized ordinary CLI') +module._initialize_cli = forbidden +sys.argv = ['foreign-program', 'doctor'] +assert module.main() == 'passive' +""" + _run_import_check(tmp_path, code, completion) + + +@pytest.mark.parametrize("entry", ["amplifier_app_cli.__main__", "amplifier_app_cli.main"]) +def test_cached_entry_suppresses_further_bytecode(entry, tmp_path): + code = """ +import importlib +import sys +entry_name = sys.argv[1] +sys.argv = ['foreign-program', 'doctor'] +entry = importlib.import_module(entry_name) +doctor = importlib.import_module('amplifier_app_cli.commands.doctor') +doctor.main = lambda: 'passive' +sys.dont_write_bytecode = False +assert entry.main() == 'passive' +assert sys.dont_write_bytecode +assert 'amplifier_foundation' not in sys.modules +""" + _run_import_check(tmp_path, code, entry) \ No newline at end of file diff --git a/tests/test_shell_completion.py b/tests/test_shell_completion.py index dd94e2b9..582abfb1 100644 --- a/tests/test_shell_completion.py +++ b/tests/test_shell_completion.py @@ -342,6 +342,13 @@ def test_actual_cli_protocol_emits_native_dynamic_records( ("words", "cword", "expected_records"), [ ("amplifier bun", "1", {"plain,bundle"}), + ("amplifier doc", "1", {"plain,doctor"}), + ( + "amplifier doctor --", + "2", + {"plain,--format", "plain,--session", "plain,--help"}, + ), + ("amplifier doctor --format ", "3", {"plain,text", "plain,json"}), ( "amplifier run --", "2", From 3b92f17fea1f4ce4c634b2f35c3be465950292eb Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:52:43 -0700 Subject: [PATCH 2/2] test: keep doctor parameter IDs within Windows limits Generated with Amplifier Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- tests/test_doctor.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/test_doctor.py b/tests/test_doctor.py index e1119676..5f2bc646 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -73,6 +73,10 @@ def test_missing_files_complete_without_creating_home(target): ("a: " + "[" * 34 + "0" + "]" * 34, "too_large"), ("a: [" + ",".join("0" for _ in range(10001)) + "]", "too_large"), ("#" * (diag.SETTINGS_LIMIT + 1), "too_large"), +], ids=[ + "empty", "empty-mapping", "null", "list-root", "boolean-root", + "malformed", "duplicate-key", "alias", "standard-tag", "custom-tag", + "unhashable-key", "scalar-limit", "depth-limit", "node-limit", "file-limit", ]) def test_bounded_settings(target, text, status): write_settings(target, text)