Skip to content

perf(cli): load heartbeat prompt on demand #7704

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huangruiteng merged 2 commits into from
Oct 30, 2026
Merged

perf(cli): load heartbeat prompt on demand #7704

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
75c6d15
perf(cli): load heartbeat prompt on demand
cocolord Oct 29, 2026
4f9a2e5
test(cli): qualify heartbeat loading and command ownership
cocolord Oct 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions examples/cli-support-control-command-modularization-smoke.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand All @@ -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": (
Expand Down
231 changes: 1 addition & 230 deletions loopx/cli_commands/support_control.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -91,7 +74,6 @@
"dashboard",
"dsh-launch",
"heartbeat-prequota",
"heartbeat-prompt",
"promotion-gate",
"promotion-readiness",
"upgrade-plan",
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading