diff --git a/examples/cli-support-control-command-modularization-smoke.py b/examples/cli-support-control-command-modularization-smoke.py index 051cc789fa..8866d63072 100644 --- a/examples/cli-support-control-command-modularization-smoke.py +++ b/examples/cli-support-control-command-modularization-smoke.py @@ -9,12 +9,13 @@ ROOT = Path(__file__).resolve().parents[1] CLI = ROOT / "loopx" / "cli.py" MODULE = ROOT / "loopx" / "cli_commands" / "support_control.py" +HEARTBEAT = MODULE.with_name("support_control_heartbeat.py") INIT = ROOT / "loopx" / "cli_commands" / "__init__.py" def run_cli(*args: str) -> subprocess.CompletedProcess[str]: return subprocess.run( - [sys.executable, "-m", "loopx.cli", *args], + [sys.executable, "-m", "loopx.entrypoint", *args], cwd=ROOT, text=True, capture_output=True, @@ -77,10 +78,8 @@ def main() -> None: "SUPPORT_CONTROL_COMMANDS", "register_support_control_commands", "handle_support_control_command", - "resolve_heartbeat_active_state", "backup-state", "chat", - "heartbeat-prompt", "promotion-gate", "upgrade-plan", "update", @@ -92,6 +91,10 @@ def main() -> None: require("handle_support_control_command" in cli_source, "cli.py did not dispatch support/control commands") require("register_support_control_commands" in init_source, "__init__ omitted support/control registration") require("handle_support_control_command" in init_source, "__init__ omitted support/control handler") + heartbeat_source = HEARTBEAT.read_text(encoding="utf-8") + require("build_heartbeat_prompt(" not in module_source, "support/control retained duplicate heartbeat dispatch") + for marker in ("handle_heartbeat_prompt_command", "resolve_heartbeat_active_state", "build_heartbeat_prompt("): + require(marker in heartbeat_source, f"heartbeat handler missing {marker}") for command, options in { "backup-state": ( diff --git a/loopx/cli_commands/support_control.py b/loopx/cli_commands/support_control.py index d0fed0eeef..38f4641701 100644 --- a/loopx/cli_commands/support_control.py +++ b/loopx/cli_commands/support_control.py @@ -4,31 +4,17 @@ from collections.abc import Callable from pathlib import Path -from ..agent_registry import ( - agent_profile_from_registry, - load_goal_from_registry, - registered_agent_ids_from_registry, - require_registered_agent_id, -) from ..chat_server import ( DEFAULT_CHAT_HOST, DEFAULT_CHAT_PORT, serve_chat, ) -from ..control_plane.scheduler.execution_context import SchedulerRuntimeProfile from ..dashboard_launcher import launch_dashboard from ..runtime_activation import replace_managed_loopx_chat as replace_existing_loopx_chat -from ..execution_profile import execution_profile_turn_granularity from ..heartbeat_prequota import ( render_heartbeat_pre_quota_markdown, run_heartbeat_pre_quota, ) -from ..heartbeat_prompt import ( - build_heartbeat_prompt, - build_heartbeat_prompt_error_payload, - project_heartbeat_agent_input, - render_heartbeat_prompt_markdown, -) from ..kiro_cli_goal_mode import KIRO_CLI_BIN from ..paths import default_public_scan_root from ..registry import ( @@ -59,10 +45,7 @@ handle_promotion_control_command, register_promotion_control_commands, ) -from .support_control_registry import ( - explicit_global_registry, - resolve_heartbeat_active_state, -) +from .support_control_registry import explicit_global_registry from .support_control_supervisor import ( SUPERVISOR_CONTROL_COMMANDS, handle_supervisor_control_command, @@ -91,7 +74,6 @@ "dashboard", "dsh-launch", "heartbeat-prequota", - "heartbeat-prompt", "promotion-gate", "promotion-readiness", "upgrade-plan", @@ -287,217 +269,6 @@ def handle_support_control_command( ) return 0 - if args.command == "heartbeat-prompt": - active_state = None - resolved_active_state = None - active_state_source = None - registered_agents = None - effective_agent_id = args.agent_id - agent_registry_path = registry_path - requested_runtime_profile = ( - SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT.value - if args.codex_app - else SchedulerRuntimeProfile.TRAE_APP.value - if getattr(args, "trae_app", False) - else args.runtime_profile - ) - try: - active_state, resolved_active_state, active_state_source = ( - resolve_heartbeat_active_state( - goal_id=args.goal_id, - active_state_arg=args.active_state, - registry_path=registry_path, - runtime_root_arg=args.runtime_root, - allow_global_goal_lookup_fallback=not registry_was_supplied, - ) - ) - agent_registry_path = registry_path - if active_state_source.startswith("registry:"): - agent_registry_path = Path( - active_state_source.removeprefix("registry:") - ) - registered_agents = registered_agent_ids_from_registry( - agent_registry_path, args.goal_id - ) - registry_goal = load_goal_from_registry( - agent_registry_path, - args.goal_id, - ) - turn_granularity = execution_profile_turn_granularity( - registry_goal.get("execution_profile") - if isinstance(registry_goal, dict) - else None - ) - agent_profile = None - if args.agent_id: - effective_agent_id = require_registered_agent_id( - registry_path=agent_registry_path, - goal_id=args.goal_id, - agent_id=args.agent_id, - field="agent_id", - ) - agent_profile = agent_profile_from_registry( - agent_registry_path, args.goal_id, effective_agent_id - ) - reward_memory_enabled = False - if effective_agent_id is not None: - from ..capabilities.reward_memory.configuration import ( - reward_memory_automatic_ingest_for_agent, - ) - - reward_memory_enabled = reward_memory_automatic_ingest_for_agent( - registry_goal if isinstance(registry_goal, dict) else {}, - effective_agent_id, - ) - explicit_scheduler_fields = ( - args.host_surface, - args.scheduler_owner, - args.execution_mode, - ) - app_alias_count = int(bool(args.codex_app)) + int( - bool(getattr(args, "trae_app", False)) - ) - if app_alias_count > 1: - raise ValueError( - "--codex-app and --trae_app are mutually exclusive" - ) - if app_alias_count and ( - args.runtime_profile or any(explicit_scheduler_fields) - ): - raise ValueError( - "app runtime aliases cannot be combined with --runtime-profile, " - "--host-surface, --scheduler-owner, or --execution-mode" - ) - if args.runtime_profile and any(explicit_scheduler_fields): - raise ValueError( - "--runtime-profile cannot be combined with --host-surface, " - "--scheduler-owner, or --execution-mode" - ) - runtime_profile = requested_runtime_profile - from copy import copy - from .context_machine import apply_context_machine_options - - # Persist caller choices only. Machine defaults and private paths - # must resolve again on the next wake, including after rollback. - bootstrap_args = copy(args) - apply_context_machine_options(args, registry_path=agent_registry_path, - runtime_root_arg=args.runtime_root, heartbeat=True) - payload = build_heartbeat_prompt( - goal_id=args.goal_id, - active_state=active_state, - active_state_source=active_state_source, - resolved_active_state=resolved_active_state, - material_queue_rule=args.material_rule, - permission_rule=args.permission_rule, - full=bool(args.full), - compact=bool(args.compact), - brief=bool(args.brief), - thin=bool(args.thin), - cli_bin=args.cli_bin, - runtime_root=args.runtime_root, - registry_path=agent_registry_path.resolve(), - agent_id=effective_agent_id, - agent_scopes=args.agent_scopes, - agent_profile=agent_profile, - registered_agents=registered_agents, - available_capabilities=args.available_capabilities, - runtime_profile=runtime_profile, - scheduler_execution_context=( - { - "host_surface": args.host_surface, - "scheduler_owner": args.scheduler_owner, - "execution_mode": args.execution_mode, - } - if any(explicit_scheduler_fields) - else None - ), - visible_goal_host=args.visible_goal_host, - turn_granularity=turn_granularity, - turn_instance_id=args.turn_instance_id, - decision_output_root=args.decision_output_root, - decision_handoff_socket=args.decision_handoff_socket, - codex_context_dir=args.codex_context_dir, - codex_context_root=args.codex_context_root, - context_delivery=args.context_delivery, - turn_envelope=args.turn_envelope, - reward_memory_enabled=reward_memory_enabled, - ) - if args.bootstrap and payload.get("ok"): - from ..control_plane.heartbeat.bootstrap_prompt import goal_bootstrap - from ..control_plane.heartbeat.budget import build_interface_budget - body = goal_bootstrap(bootstrap_args, registry=agent_registry_path) - payload["task_body"] = body - payload["bootstrap"] = True - payload["interface_budget"] = build_interface_budget( - task_body=body, goal_id=args.goal_id, - active_state=str(payload.get("active_state") or ""), thin=True, - registry_path=str(agent_registry_path.resolve()), - runtime_root=(str(Path(args.runtime_root).expanduser().resolve()) - if args.runtime_root else None), - ) - if not payload["interface_budget"]["within_budget"]: - raise ValueError("bootstrap exceeds the thin budget; move lengthy policy into registered state") - except Exception as exc: - fallback_active_state = active_state - fallback_resolved_active_state = resolved_active_state - fallback_active_state_source = active_state_source - if fallback_active_state is None and args.active_state: - fallback_active_state = Path(args.active_state).expanduser() - fallback_resolved_active_state = ( - fallback_resolved_active_state or fallback_active_state - ) - fallback_active_state_source = ( - fallback_active_state_source or "explicit" - ) - elif fallback_active_state_source is None: - fallback_active_state_source = "registry" - payload = build_heartbeat_prompt_error_payload( - goal_id=args.goal_id, - error=str(exc), - active_state=fallback_active_state, - active_state_source=fallback_active_state_source, - resolved_active_state=fallback_resolved_active_state, - material_queue_rule=args.material_rule, - permission_rule=args.permission_rule, - full=bool(args.full), - compact=bool(args.compact), - brief=bool(args.brief), - thin=bool(args.thin), - cli_bin=args.cli_bin, - runtime_root=args.runtime_root, - registry_path=agent_registry_path.resolve(), - agent_id=effective_agent_id or args.agent_id, - agent_scopes=args.agent_scopes, - registered_agents=registered_agents, - available_capabilities=args.available_capabilities, - ) - selected_output_format = output_format(args) - recurring_runtime_profile = requested_runtime_profile in { - None, - SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT.value, - SchedulerRuntimeProfile.TRAE_APP.value, - SchedulerRuntimeProfile.GENERIC_CLI_AGENT_LOOP.value, - # Kiro CLI moved off generic_cli onto its own profile; its facade - # still reads the thin Agent input, so keep that output shape. - SchedulerRuntimeProfile.KIRO_CLI_VISIBLE.value, - } - recurring_thin_surface = bool( - recurring_runtime_profile and args.visible_goal_host is None - ) - output_payload = ( - project_heartbeat_agent_input(payload) - if selected_output_format == "json" - and payload.get("thin") is True - and recurring_thin_surface - else payload - ) - print_payload( - output_payload, - selected_output_format, - render_heartbeat_prompt_markdown, - ) - return 0 if payload.get("ok") else 1 - supervisor_result = handle_supervisor_control_command( args, registry_path=registry_path, diff --git a/loopx/cli_commands/support_control_heartbeat.py b/loopx/cli_commands/support_control_heartbeat.py new file mode 100644 index 0000000000..ffb47f3065 --- /dev/null +++ b/loopx/cli_commands/support_control_heartbeat.py @@ -0,0 +1,246 @@ +from __future__ import annotations + +import argparse +from collections.abc import Callable +from pathlib import Path + +from ..agent_registry import ( + agent_profile_from_registry, + load_goal_from_registry, + registered_agent_ids_from_registry, + require_registered_agent_id, +) +from ..control_plane.scheduler.execution_context import SchedulerRuntimeProfile +from ..execution_profile import execution_profile_turn_granularity +from ..heartbeat_prompt import ( + build_heartbeat_prompt, + build_heartbeat_prompt_error_payload, + project_heartbeat_agent_input, + render_heartbeat_prompt_markdown, +) +from .support_control_registry import resolve_heartbeat_active_state + +PrintPayload = Callable[ + [dict[str, object], str, Callable[[dict[str, object]], str]], + None, +] +FormatSelector = Callable[..., str] + + +def handle_heartbeat_prompt_command( + args: argparse.Namespace, + *, + registry_path: Path, + registry_was_supplied: bool, + print_payload: PrintPayload, + output_format: FormatSelector, +) -> int: + active_state = None + resolved_active_state = None + active_state_source = None + registered_agents = None + effective_agent_id = args.agent_id + agent_registry_path = registry_path + requested_runtime_profile = ( + SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT.value + if args.codex_app + else SchedulerRuntimeProfile.TRAE_APP.value + if getattr(args, "trae_app", False) + else args.runtime_profile + ) + try: + active_state, resolved_active_state, active_state_source = ( + resolve_heartbeat_active_state( + goal_id=args.goal_id, + active_state_arg=args.active_state, + registry_path=registry_path, + runtime_root_arg=args.runtime_root, + allow_global_goal_lookup_fallback=not registry_was_supplied, + ) + ) + agent_registry_path = registry_path + if active_state_source.startswith("registry:"): + agent_registry_path = Path( + active_state_source.removeprefix("registry:") + ) + registered_agents = registered_agent_ids_from_registry( + agent_registry_path, args.goal_id + ) + registry_goal = load_goal_from_registry( + agent_registry_path, + args.goal_id, + ) + turn_granularity = execution_profile_turn_granularity( + registry_goal.get("execution_profile") + if isinstance(registry_goal, dict) + else None + ) + agent_profile = None + if args.agent_id: + effective_agent_id = require_registered_agent_id( + registry_path=agent_registry_path, + goal_id=args.goal_id, + agent_id=args.agent_id, + field="agent_id", + ) + agent_profile = agent_profile_from_registry( + agent_registry_path, args.goal_id, effective_agent_id + ) + reward_memory_enabled = False + if effective_agent_id is not None: + from ..capabilities.reward_memory.configuration import ( + reward_memory_automatic_ingest_for_agent, + ) + + reward_memory_enabled = reward_memory_automatic_ingest_for_agent( + registry_goal if isinstance(registry_goal, dict) else {}, + effective_agent_id, + ) + explicit_scheduler_fields = ( + args.host_surface, + args.scheduler_owner, + args.execution_mode, + ) + app_alias_count = int(bool(args.codex_app)) + int( + bool(getattr(args, "trae_app", False)) + ) + if app_alias_count > 1: + raise ValueError( + "--codex-app and --trae_app are mutually exclusive" + ) + if app_alias_count and ( + args.runtime_profile or any(explicit_scheduler_fields) + ): + raise ValueError( + "app runtime aliases cannot be combined with --runtime-profile, " + "--host-surface, --scheduler-owner, or --execution-mode" + ) + if args.runtime_profile and any(explicit_scheduler_fields): + raise ValueError( + "--runtime-profile cannot be combined with --host-surface, " + "--scheduler-owner, or --execution-mode" + ) + runtime_profile = requested_runtime_profile + from copy import copy + from .context_machine import apply_context_machine_options + + # Persist caller choices only. Machine defaults and private paths + # must resolve again on the next wake, including after rollback. + bootstrap_args = copy(args) + apply_context_machine_options(args, registry_path=agent_registry_path, + runtime_root_arg=args.runtime_root, heartbeat=True) + payload = build_heartbeat_prompt( + goal_id=args.goal_id, + active_state=active_state, + active_state_source=active_state_source, + resolved_active_state=resolved_active_state, + material_queue_rule=args.material_rule, + permission_rule=args.permission_rule, + full=bool(args.full), + compact=bool(args.compact), + brief=bool(args.brief), + thin=bool(args.thin), + cli_bin=args.cli_bin, + runtime_root=args.runtime_root, + registry_path=agent_registry_path.resolve(), + agent_id=effective_agent_id, + agent_scopes=args.agent_scopes, + agent_profile=agent_profile, + registered_agents=registered_agents, + available_capabilities=args.available_capabilities, + runtime_profile=runtime_profile, + scheduler_execution_context=( + { + "host_surface": args.host_surface, + "scheduler_owner": args.scheduler_owner, + "execution_mode": args.execution_mode, + } + if any(explicit_scheduler_fields) + else None + ), + visible_goal_host=args.visible_goal_host, + turn_granularity=turn_granularity, + turn_instance_id=args.turn_instance_id, + decision_output_root=args.decision_output_root, + decision_handoff_socket=args.decision_handoff_socket, + codex_context_dir=args.codex_context_dir, + codex_context_root=args.codex_context_root, + context_delivery=args.context_delivery, + turn_envelope=args.turn_envelope, + reward_memory_enabled=reward_memory_enabled, + ) + if args.bootstrap and payload.get("ok"): + from ..control_plane.heartbeat.bootstrap_prompt import goal_bootstrap + from ..control_plane.heartbeat.budget import build_interface_budget + body = goal_bootstrap(bootstrap_args, registry=agent_registry_path) + payload["task_body"] = body + payload["bootstrap"] = True + payload["interface_budget"] = build_interface_budget( + task_body=body, goal_id=args.goal_id, + active_state=str(payload.get("active_state") or ""), thin=True, + registry_path=str(agent_registry_path.resolve()), + runtime_root=(str(Path(args.runtime_root).expanduser().resolve()) + if args.runtime_root else None), + ) + if not payload["interface_budget"]["within_budget"]: + raise ValueError("bootstrap exceeds the thin budget; move lengthy policy into registered state") + except Exception as exc: + fallback_active_state = active_state + fallback_resolved_active_state = resolved_active_state + fallback_active_state_source = active_state_source + if fallback_active_state is None and args.active_state: + fallback_active_state = Path(args.active_state).expanduser() + fallback_resolved_active_state = ( + fallback_resolved_active_state or fallback_active_state + ) + fallback_active_state_source = ( + fallback_active_state_source or "explicit" + ) + elif fallback_active_state_source is None: + fallback_active_state_source = "registry" + payload = build_heartbeat_prompt_error_payload( + goal_id=args.goal_id, + error=str(exc), + active_state=fallback_active_state, + active_state_source=fallback_active_state_source, + resolved_active_state=fallback_resolved_active_state, + material_queue_rule=args.material_rule, + permission_rule=args.permission_rule, + full=bool(args.full), + compact=bool(args.compact), + brief=bool(args.brief), + thin=bool(args.thin), + cli_bin=args.cli_bin, + runtime_root=args.runtime_root, + registry_path=agent_registry_path.resolve(), + agent_id=effective_agent_id or args.agent_id, + agent_scopes=args.agent_scopes, + registered_agents=registered_agents, + available_capabilities=args.available_capabilities, + ) + selected_output_format = output_format(args) + recurring_runtime_profile = requested_runtime_profile in { + None, + SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT.value, + SchedulerRuntimeProfile.TRAE_APP.value, + SchedulerRuntimeProfile.GENERIC_CLI_AGENT_LOOP.value, + # Kiro CLI moved off generic_cli onto its own profile; its facade + # still reads the thin Agent input, so keep that output shape. + SchedulerRuntimeProfile.KIRO_CLI_VISIBLE.value, + } + recurring_thin_surface = bool( + recurring_runtime_profile and args.visible_goal_host is None + ) + output_payload = ( + project_heartbeat_agent_input(payload) + if selected_output_format == "json" + and payload.get("thin") is True + and recurring_thin_surface + else payload + ) + print_payload( + output_payload, + selected_output_format, + render_heartbeat_prompt_markdown, + ) + return 0 if payload.get("ok") else 1 diff --git a/loopx/cli_runtime.py b/loopx/cli_runtime.py index 5909da9d18..3f243524c3 100644 --- a/loopx/cli_runtime.py +++ b/loopx/cli_runtime.py @@ -77,6 +77,7 @@ "agent-context", "agent-directory", "agent-capabilities", "handoff", "handoff-mode", "performance-diagnosis", + "heartbeat-prompt", } @@ -394,6 +395,10 @@ def _build_selected_parser(command: str) -> LoopXArgumentParser: from .capabilities.performance_diagnosis.cli import register_performance_diagnosis_commands register_performance_diagnosis_commands(subparsers, add_subcommand_format) + elif command == "heartbeat-prompt": + from .cli_commands.support_control_heartbeat_registration import register_heartbeat_control_commands + + register_heartbeat_control_commands(subparsers, add_subcommand_format) else: # pragma: no cover - caller guards the private interface raise ValueError(f"unsupported selected command: {command}") return parser @@ -405,6 +410,14 @@ def _dispatch_common_command( registry_path: Path, allow_missing_registry: bool, ) -> int | None: + if args.command == "heartbeat-prompt": + from .cli_commands.support_control_heartbeat import handle_heartbeat_prompt_command + + return handle_heartbeat_prompt_command( + args, registry_path=registry_path, + registry_was_supplied=not allow_missing_registry, + print_payload=print_payload, output_format=output_format, + ) if args.command == "performance-diagnosis": from .capabilities.performance_diagnosis.cli import handle_performance_diagnosis_command diff --git a/tests/control_plane/test_heartbeat_agent_input.py b/tests/control_plane/test_heartbeat_agent_input.py index 3d81a62b17..8782181612 100644 --- a/tests/control_plane/test_heartbeat_agent_input.py +++ b/tests/control_plane/test_heartbeat_agent_input.py @@ -56,7 +56,7 @@ def test_thin_cli_preserves_readable_admission_and_authority_with_host_scope(tmp "state_file": "state.md", "coordination": {"registered_agents": agents}}], }), encoding="utf-8") - command = [sys.executable, "-m", "loopx.cli", "--format", "json", + command = [sys.executable, "-m", "loopx.entrypoint", "--format", "json", "--registry", str(registry), "heartbeat-prompt", "--goal-id", "synthetic-goal-validationx", "--agent-id", "worker-a", "--codex-app", "--thin", "--agent-scope", scope] diff --git a/tests/control_plane/test_heartbeat_cli.py b/tests/control_plane/test_heartbeat_cli.py new file mode 100644 index 0000000000..b83d10cd34 --- /dev/null +++ b/tests/control_plane/test_heartbeat_cli.py @@ -0,0 +1,143 @@ +"""Heartbeat CLI routing preserves the public grammar, context and recovery.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from test_cli_entrypoint import run_cli_batch, run_cli_main, run_isolated_script, write_command_fixture + + +PARSER_CASES = [ + (["heartbeat-prompt", "--help"], 0), + (["--format=json", "heartbeat-prompt", "--help"], 0), + (["heartbeat-prompt"], 2), + (["heartbeat-prompt", "--goal-id"], 2), + (["heartbeat-prompt", "--goal", "fixture"], 2), + (["heartbeat-prompt", "--goal-id=fixture", "--unknown"], 2), + (["heartbeat-prompt", "--goal-id=fixture", "--full", "--thin"], 2), + (["heartbeat-prompt", "--goal-id=fixture", "--runtime-profile=unknown"], 2), + (["heartbeat-prompt", "--goal-id=fixture", "--context-delivery=invalid"], 2), + (["heartbeat-prompt", "--goal-id=fixture", "--codex-context-dir=a", "--codex-context-root=b"], 2), + (["heartbeat-prompt", "--goal-id=fixture", "--format=invalid"], 2), + (["--format=invalid", "heartbeat-prompt", "--help"], 2), +] + + +def test_heartbeat_parser_and_full_diagnostics_match(): + cases = [args for args, _ in PARSER_CASES] + full = run_cli_batch("loopx.cli", cases) + entry = run_cli_batch("loopx.entrypoint", cases) + assert [row["returncode"] for row in full] == [code for _, code in PARSER_CASES] + assert entry == full + + +def heartbeat_args(registry: Path, runtime: Path) -> list[str]: + return ["--registry", str(registry), "--runtime-root", str(runtime), + "heartbeat-prompt", "--goal-id=perf-goal", "--agent-id=perf-agent"] + + +@pytest.mark.parametrize("global_format,local_format", [("json", None), ("markdown", None), ("markdown", "json")]) +def test_heartbeat_output_and_format_precedence(tmp_path, global_format, local_format): + registry, runtime = write_command_fixture(tmp_path) + base = ["--format", global_format, *heartbeat_args(registry, runtime), "--codex-app"] + if local_format: + base += ["--format", local_format] + cases = [base + mode for mode in ([], ["--thin"], ["--brief"], ["--compact"], ["--full"], ["--bootstrap"])] + full = run_cli_batch("loopx.cli", cases) + entry = run_cli_batch("loopx.entrypoint", cases) + assert entry == full + assert all(row["returncode"] == 0 for row in entry) + if (local_format or global_format) == "json": + for row in entry: + payload = json.loads(row["stdout"]) + assert payload["ok"] is True + assert payload["agent_id"] == "perf-agent" + assert "--registry" in payload["task_body"] + assert str(registry) in payload["task_body"] + assert "--runtime-root" in payload["task_body"] + + +def test_heartbeat_semantic_errors_keep_recovery_and_exit_status(tmp_path): + registry, runtime = write_command_fixture(tmp_path) + base = ["--format=json", *heartbeat_args(registry, runtime)] + cases = [base + options for options in ( + ["--codex-app", "--trae_app"], + ["--codex-app", "--runtime-profile=codex_app_heartbeat"], + ["--runtime-profile=codex_app_heartbeat", "--host-surface=codex_app"], + ["--agent-id=unregistered"], + ["--goal-id=missing"], + ["--context-delivery=on", "--trae_app"], + ["--turn-envelope"], + ["--decision-handoff-socket=missing"], + )] + full = run_cli_batch("loopx.cli", cases) + entry = run_cli_batch("loopx.entrypoint", cases) + assert entry == full + assert all(row["returncode"] == 1 for row in entry) + for row in entry: + payload = json.loads(row["stdout"]) + assert payload["ok"] is False + assert payload.get("error") or payload.get("task_body") + + +def test_heartbeat_explicit_missing_registry_retains_fallback_contract(tmp_path): + registry, runtime = write_command_fixture(tmp_path) + # Explicitly missing registry and an absent Goal have different recovery; + # compare both without supplying a real user's global registry. + cases = [["--format=json", *heartbeat_args(registry.with_name("absent.json"), runtime)], + ["--format=json", *heartbeat_args(registry, runtime), "--goal-id=absent"]] + assert run_cli_batch("loopx.entrypoint", cases) == run_cli_batch("loopx.cli", cases) + assert not registry.with_name("absent.json").exists() + + +def test_heartbeat_read_only_prompt_without_node(tmp_path): + registry, runtime = write_command_fixture(tmp_path) + args = ["--format=json", *heartbeat_args(registry, runtime), "--codex-app", "--context-delivery=off"] + env = {"PATH": str(tmp_path / "no-executables")} + full = run_cli_main("loopx.cli", args, environment=env) + entry = run_cli_main("loopx.entrypoint", args, environment=env) + assert (entry.returncode, entry.stdout, entry.stderr) == (full.returncode, full.stdout, full.stderr) + assert entry.returncode == 0, entry.stderr + assert json.loads(entry.stdout)["ok"] is True + + +@pytest.mark.parametrize("mode", ["help", "thin", "bootstrap"]) +def test_cold_heartbeat_avoids_unrelated_modules_and_external_io(tmp_path, mode): + registry, runtime = write_command_fixture(tmp_path) + args = (["heartbeat-prompt", "--help"] if mode == "help" else + ["--format=json", *heartbeat_args(registry, runtime), "--codex-app", + "--context-delivery=off", "--" + mode]) + script = f""" +import contextlib +import io +import json +import sys + +events = [] +def audit(event, args): + if event in {{"subprocess.Popen", "os.system", "os.posix_spawn"}} or event.startswith("socket."): + events.append(event) + raise AssertionError("unexpected external IO: " + event) +sys.addaudithook(audit) + +from loopx.entrypoint import main +output = io.StringIO() +with contextlib.redirect_stdout(output): + try: + code = main({args!r}) + except SystemExit as exc: + code = exc.code +assert code == 0, output.getvalue() +assert not events, events +if {mode!r} != "help": + assert json.loads(output.getvalue())["ok"] is True +for module in ("loopx.cli", "loopx.cli_commands.support_control", "loopx.chat_server", + "loopx.dashboard_launcher", "loopx.status_server"): + assert module not in sys.modules, "unexpected module import: " + module +if {mode!r} == "help": + assert "loopx.cli_commands.support_control_heartbeat" not in sys.modules +""" + result = run_isolated_script(script, environment={"LOOPX_USAGE_PING": "0"}) + assert result.returncode == 0, result.stderr diff --git a/tests/control_plane/test_heartbeat_registry_route.py b/tests/control_plane/test_heartbeat_registry_route.py index 693fdd23d7..3ac6efc2b2 100644 --- a/tests/control_plane/test_heartbeat_registry_route.py +++ b/tests/control_plane/test_heartbeat_registry_route.py @@ -121,7 +121,7 @@ def test_external_poll_opt_out_keeps_prequota_out_of_every_host_prompt(tmp_path, def _execute(argv: list[str], cwd: Path): env = dict(os.environ, PYTHONPATH=str(cli.REPO_ROOT)) - result = subprocess.run([sys.executable, "-m", "loopx.cli", "--format", "json", *argv], + result = subprocess.run([sys.executable, "-m", "loopx.entrypoint", "--format", "json", *argv], cwd=cwd, env=env, text=True, capture_output=True, check=False) assert result.stdout, result.stderr return result.returncode, json.loads(result.stdout)