Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
65 changes: 65 additions & 0 deletions examples/control_plane/heartbeat-prompt-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -1288,6 +1288,12 @@ def main() -> int:

with tempfile.TemporaryDirectory() as raw_tmp:
root = Path(raw_tmp)
long_profile_scope = (
"Investigate product and runtime reliability, coordinate bounded peer work, "
"preserve task identity, qualify evidence and acceptance, and return actionable "
"results to the original conversation without claiming another agent's authority."
)
assert 180 < len(long_profile_scope) <= 320
project = root / "project"
state_file = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md"
registry_path = project / ".loopx" / "registry.json"
Expand All @@ -1313,6 +1319,10 @@ def main() -> int:
"registered_agents": ["codex-main-control", "codex-side-bypass"],
"agent_model": "peer_v1",
"agent_profiles": {
"codex-main-control": {
"schema_version": "agent_profile_v1",
"scope_summary": long_profile_scope,
},
"codex-side-bypass": {
"schema_version": "agent_profile_v1",
"scope_summary": "productization showcase docs lane",
Expand Down Expand Up @@ -1521,6 +1531,61 @@ def main() -> int:
cli_profile_scoped_payload
)

long_profile_result = subprocess.run(
[
sys.executable,
"-m",
"loopx.cli",
"--format",
"json",
"--registry",
str(registry_path),
"heartbeat-prompt",
"--goal-id",
GOAL_ID,
"--thin",
"--agent-id",
"codex-main-control",
],
cwd=REPO_ROOT,
check=False,
capture_output=True,
text=True,
)
assert long_profile_result.returncode == 0, long_profile_result.stdout
long_profile_payload = json.loads(long_profile_result.stdout)
assert long_profile_payload["ok"] is True, long_profile_payload
assert long_profile_scope in normalized(long_profile_payload["task_body"]), long_profile_payload

explicit_scope_result = subprocess.run(
[
sys.executable,
"-m",
"loopx.cli",
"--format",
"json",
"--registry",
str(registry_path),
"heartbeat-prompt",
"--goal-id",
GOAL_ID,
"--thin",
"--agent-id",
"codex-main-control",
"--agent-scope",
"explicit bounded task",
],
cwd=REPO_ROOT,
check=False,
capture_output=True,
text=True,
)
assert explicit_scope_result.returncode == 0, explicit_scope_result.stdout
explicit_scope_payload = json.loads(explicit_scope_result.stdout)
assert explicit_scope_payload["ok"] is True, explicit_scope_payload
assert "explicit bounded task" in explicit_scope_payload["task_body"], explicit_scope_payload
assert long_profile_scope not in explicit_scope_payload["task_body"], explicit_scope_payload

cli_unknown_scoped = subprocess.run(
[
sys.executable,
Expand Down
3 changes: 2 additions & 1 deletion loopx/control_plane/agents/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
}
AGENT_PROFILE_ACTION_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_*-]{0,79}$")
AGENT_PROFILE_ACTION_PATTERN_LIMIT = 16
AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS = 320
AGENT_PROFILE_HIERARCHY_ROLES = {
"leader",
"manager",
Expand Down Expand Up @@ -173,7 +174,7 @@ def normalize_agent_profile(
"scope_summary": _bounded_text(
raw_profile.get("scope_summary"),
field="scope_summary",
limit=320,
limit=AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS,
),
"default_task_classes": task_classes,
"vision_requirement": _vision_requirement(
Expand Down
8 changes: 7 additions & 1 deletion loopx/control_plane/heartbeat/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,20 @@
import shlex
from typing import Any

from ..agents.profile import AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS
from ..agents.runtime_model import PEER_AGENT_PROFILE_SCHEMA_VERSION


def normalize_agent_scope(value: Any) -> str | None:
candidate = " ".join(str(value or "").strip().split())
if not candidate:
return None
if len(candidate) > 180 or any(char in candidate for char in "<>"):
if len(candidate) > AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS:
raise ValueError(
"agent scope must be at most "
f"{AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS} characters"
)
if any(char in candidate for char in "<>"):
raise ValueError("agent scope must be compact text without angle brackets")
return candidate

Expand Down
9 changes: 6 additions & 3 deletions loopx/control_plane/heartbeat/builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -347,12 +347,15 @@ def build_heartbeat_prompt(
scheduler_execution_context=scheduler_execution_context,
)
explicit_agent_scopes = normalize_agent_scopes(agent_scopes)
profile_agent_scopes = agent_profile_scopes(agent_profile)
normalized_agent_scopes = explicit_agent_scopes or profile_agent_scopes
if explicit_agent_scopes:
normalized_agent_scopes = explicit_agent_scopes
agent_scope_source = "argument"
else:
normalized_agent_scopes = agent_profile_scopes(agent_profile)
agent_scope_source = "agent_profile_v1" if normalized_agent_scopes else None
if traex_visible_goal:
for scope in normalized_agent_scopes:
validate_visible_goal_policy_rule(field="agent_scope", value=scope)
agent_scope_source = "argument" if explicit_agent_scopes else "agent_profile_v1" if profile_agent_scopes else None
if normalized_agent_scopes and not normalized_agent_id:
raise ValueError("--agent-scope requires --agent-id so claimed_by uses a registered agent")
normalized_registered_agents = normalize_registered_agents(registered_agents)
Expand Down
59 changes: 59 additions & 0 deletions tests/control_plane/test_heartbeat_prompt_support.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,10 @@

import pytest

from loopx.control_plane.agents.profile import (
AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS,
normalize_agent_profile,
)
from loopx.control_plane.heartbeat.agent import (
agent_prompt_command_args,
agent_profile_scopes,
Expand Down Expand Up @@ -112,6 +116,61 @@ def test_agent_profile_scopes_reads_common_profile_keys() -> None:
assert agent_profile_scopes(profile) == ["one", "two three"]


def test_heartbeat_accepts_the_full_authored_profile_scope() -> None:
agent_id = "codex-product-capability"
scope = "x" * AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS
profile = normalize_agent_profile(
{"agent_id": agent_id, "scope_summary": scope},
registered_agents=[agent_id],
)
payload = build_heartbeat_prompt(
goal_id="profile-scope-limit",
thin=True,
agent_id=agent_id,
registered_agents=[agent_id],
agent_profile=profile,
)
assert payload["ok"] is True
assert payload["agent_scopes"] == [scope]
assert payload["agent_scope_source"] == "agent_profile_v1"
assert scope in payload["task_body"]

explicit = build_heartbeat_prompt(
goal_id="profile-scope-limit",
thin=True,
agent_id=agent_id,
agent_scopes=["explicit bounded task"],
registered_agents=[agent_id],
agent_profile=profile,
)
assert explicit["agent_scopes"] == ["explicit bounded task"]
assert explicit["agent_scope_source"] == "argument"
assert "explicit bounded task" in explicit["task_body"]

# Explicit scope must not validate an unused profile fallback first.
legacy_profile = {"scope_summary": "x" * (AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS + 1)}
explicit_with_legacy_profile = build_heartbeat_prompt(
goal_id="profile-scope-limit",
thin=True,
agent_id=agent_id,
agent_scopes=["explicit bounded task"],
registered_agents=[agent_id],
agent_profile=legacy_profile,
)
assert explicit_with_legacy_profile["agent_scopes"] == ["explicit bounded task"]


def test_heartbeat_scope_over_profile_limit_is_rejected() -> None:
over_limit = "x" * (AGENT_PROFILE_SCOPE_SUMMARY_MAX_CHARS + 1)
with pytest.raises(ValueError, match="at most 320 characters"):
normalize_agent_scopes([over_limit])
with pytest.raises(ValueError, match="at most 320 characters"):
normalize_agent_profile(
{"agent_id": "peer-a", "scope_summary": over_limit},
registered_agents=["peer-a"],
)


def test_agent_prompt_command_args_quotes_scopes() -> None:
args = agent_prompt_command_args(
agent_id="codex-main-control",
Expand Down
Loading