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
8 changes: 8 additions & 0 deletions docs/architecture/rfcs/automatic-execution-admission-v0.md
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,14 @@ protection against clock manipulation.

For App, show desired versus observed schedule and apply failure. Unsupported
intervals require holding the affected automation, not shortening the interval.
Codex App ACK commands must re-read the matched heartbeat after the host update.
The requested RRULE and the hint's earlier observation cannot certify that update.
The typed scheduler commit rejects missing or mismatched current observations
before migration, identity reset or persistence. This applies to the console's
native follow-up and the Python compatibility adapter; failure commands retain
their original target/observed pair. Real App API updates and readback on
disposable File/SQLite Goals qualify this schedule boundary separately from
elapsed timer wakeups, installed-runtime adoption and pre-model admission.
The activation path must be qualified separately before advertising one-click
floor-safe activation. An in-model guard can stop further work after waking;
it cannot recover already consumed model tokens. Merely changing a prompt is
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,14 @@
import subprocess
import sys
import tempfile
from unittest.mock import patch


REPO_ROOT = Path(__file__).resolve().parents[2]
AGENT_ID = "codex-scheduler-route-smoke"
GOAL_ID = "scheduler-route-smoke"
sys.path.insert(0, str(REPO_ROOT))
from examples.control_plane.quota_plan_fixtures import write_codex_heartbeat_fixture # noqa: E402


def write_registry(path: Path, *, project: Path, runtime: Path) -> None:
Expand Down Expand Up @@ -74,8 +77,11 @@ def run_cli(*args: str, cwd: Path) -> dict:


def main() -> None:
with tempfile.TemporaryDirectory(prefix="loopx-scheduler-registry-route-") as tmp:
with tempfile.TemporaryDirectory(prefix="loopx-scheduler-registry-route-") as tmp, patch.dict(os.environ):
root = Path(tmp)
codex_home = root / "codex-home"
os.environ.update(CODEX_HOME=str(codex_home), CODEX_THREAD_ID="scheduler-fixture")
write_codex_heartbeat_fixture(codex_home, goal_id=GOAL_ID, agent_id=AGENT_ID, rrule="FREQ=MINUTELY;INTERVAL=3")
project = root / "project"
project.mkdir()
project_registry = project / ".loopx" / "registry.json"
Expand Down
9 changes: 8 additions & 1 deletion examples/control_plane/quota-scheduler-state-ack-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,10 @@
from __future__ import annotations

from pathlib import Path
import os
import sys
import tempfile
from unittest.mock import patch


REPO_ROOT = Path(__file__).resolve().parents[2]
Expand All @@ -15,6 +17,7 @@
from examples.control_plane.quota_plan_fixtures import ( # noqa: E402
SCOPED_AGENT_ID,
write_cli_fixture,
write_codex_heartbeat_fixture,
)
from loopx.control_plane.testing.canary_harness import run_json_cli # noqa: E402

Expand Down Expand Up @@ -45,7 +48,10 @@ def run_should_run(


def assert_cli_scheduler_ack_public_round_trip() -> None:
with tempfile.TemporaryDirectory(prefix="loopx-quota-scheduler-ack-") as tmp:
with tempfile.TemporaryDirectory(prefix="loopx-quota-scheduler-ack-") as tmp, patch.dict(os.environ):
codex_home = Path(tmp) / "codex-home"
os.environ.update(CODEX_HOME=str(codex_home), CODEX_THREAD_ID="scheduler-fixture")
write_codex_heartbeat_fixture(codex_home, goal_id=GOAL_ID, agent_id=SCOPED_AGENT_ID, rrule="FREQ=MINUTELY;INTERVAL=3")
registry_path, runtime_root, project = write_cli_fixture(
Path(tmp),
scoped_agents=True,
Expand All @@ -63,6 +69,7 @@ def assert_cli_scheduler_ack_public_round_trip() -> None:
assert ack_hint["args"]["applied_rrule"] == target_rrule, before
assert ack_hint["args"]["host_match_observed"] is True, before

write_codex_heartbeat_fixture(codex_home, goal_id=GOAL_ID, agent_id=SCOPED_AGENT_ID, rrule=target_rrule)
ack = run_json_cli(
*ack_hint["cli_args"],
registry_path=registry_path,
Expand Down
12 changes: 12 additions & 0 deletions examples/control_plane/quota_plan_fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -309,6 +309,18 @@ def append_quota_slot_spend_fixture(
)


def write_codex_heartbeat_fixture(root: Path, *, goal_id: str, agent_id: str, rrule: str) -> None:
"""Synthetic manifest for CLI readback tests; never a real App update proof."""
path = root / "automations" / "scheduler-fixture" / "automation.toml"
path.parent.mkdir(parents=True, exist_ok=True)
fields = {
"id": "scheduler-fixture", "kind": "heartbeat", "status": "ACTIVE",
"prompt": f"Advance `{goal_id}` from active state. Agent: `{agent_id}`.",
"rrule": rrule, "target_thread_id": "scheduler-fixture",
}
path.write_text("\n".join(f"{key} = {json.dumps(value)}" for key, value in fields.items()) + "\n", encoding="utf-8")


def write_cli_fixture(root: Path, *, scoped_agents: bool = False) -> tuple[Path, Path, Path]:
project = root / "project"
runtime = root / "runtime"
Expand Down
14 changes: 9 additions & 5 deletions loopx/cli_commands/quota_scheduler_followup.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,16 +146,20 @@ def build_scheduler_followup_payload(
and resolved_context.context.host_surface is HostSurface.CODEX_APP
)
if (
args.quota_command == "scheduler-fail-current"
and not observed_rrule
(
args.quota_command == "scheduler-fail-current" and not observed_rrule
or bool(getattr(args, "host_match_observed", False))
)
and codex_app_host
):
host_observation = resolve_codex_app_automation_rrule(
goal_id=args.goal_id,
agent_id=args.agent_id,
)
if host_observation.get("available") is True:
observed_rrule = str(host_observation.get("rrule") or "")
observed_rrule = (
str(host_observation.get("rrule") or "")
if host_observation.get("available") is True else ""
)

before_decision = _build_scheduler_followup_decision(
status_payload,
Expand All @@ -166,7 +170,7 @@ def build_scheduler_followup_payload(
turn_instance_id=turn_instance_id,
codex_app_current_rrule=(
args.applied_rrule
if bool(getattr(args, "host_match_observed", False))
if bool(getattr(args, "host_match_observed", False)) and not codex_app_host
else observed_rrule
),
scheduler_context=scheduler_context,
Expand Down
3 changes: 3 additions & 0 deletions loopx/control_plane/quota/scheduler_ack.py
Original file line number Diff line number Diff line change
Expand Up @@ -355,6 +355,9 @@ def record_quota_scheduler_ack_for_decision(
execute=execute,
generated_at=generated_at or now_local_iso(),
applied_rrule=applied_rrule,
observed_host_rrule=(stateful_backoff.get("host_observation") or {}).get(
"current_rrule"
),
host_match_observed=host_match_observed,
)
return _run_native_followup(
Expand Down
14 changes: 14 additions & 0 deletions loopx/control_plane/scheduler/heartbeat_commit.ts
Original file line number Diff line number Diff line change
Expand Up @@ -867,6 +867,20 @@ export async function evaluateSchedulerHeartbeatCommit(
);
}
}
if (
request.operation === "ack" && request.host_match_observed &&
(!request.observed_host_rrule || request.observed_host_rrule !== request.applied_rrule)
) {
return result(
request, path, "conflict", existing,
"host-match ACK requires a current observation matching applied_rrule",
{
reason_code: "rrule_mismatch",
applied_rrule: request.applied_rrule,
observed_host_rrule: request.observed_host_rrule,
},
);
}
if (request.execute && canonical === null) {
existing = await migrateLegacySchedulerStateUnderLock(
request.runtime_root,
Expand Down
7 changes: 5 additions & 2 deletions loopx/control_plane/scheduler/heartbeat_followup_cli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ function parseArgs(argv: string[]): ParsedArgs {
if (valueOptions.has(option)) {
const [value, next] = optionValue(argv, index, option);
index = next;
if (!value) {
if (!value && option !== "--app-automation-current-rrule") {
throw new EffectRuntimeRequestError(`${option} must not be empty`, "invalid_scheduler_followup_argv");
}
if (option === FACTS_FLAG) chunks.push(value);
Expand Down Expand Up @@ -250,7 +250,7 @@ function requestFromArgs(argv: string[]): { request: Record<string, unknown>; fo
if (expectedOperation === "host_failure" && values["--failed-rrule"]) {
matchingText(facts, "expected_rrule", values["--failed-rrule"], "--failed-rrule");
}
if (neutralObservedRrule || legacyCodexObservedRrule) {
if (expectedOperation === "host_failure" && (neutralObservedRrule || legacyCodexObservedRrule)) {
matchingText(
facts,
"observed_host_rrule",
Expand All @@ -273,6 +273,9 @@ function requestFromArgs(argv: string[]): { request: Record<string, unknown>; fo
reason_summary: values["--reason-summary"] ?? null,
host_facts: {
...facts,
...(expectedOperation === "ack" && facts.surface === "codex_app"
? { observed_host_rrule: neutralObservedRrule || legacyCodexObservedRrule }
: {}),
operation: expectedOperation,
runtime_root: parsed.runtime_root,
execute: parsed.booleans.has("--execute"),
Expand Down
26 changes: 26 additions & 0 deletions loopx/entrypoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,16 @@ def _bound_option_present(argv: list[str], option: str) -> bool:
return False


def _option_value(argv: list[str], option: str) -> str:
result = ""
for index, value in enumerate(argv):
if value.startswith(f"{option}="):
result = value.split("=", 1)[1]
if value == option and index + 1 < len(argv):
result = argv[index + 1]
return result


def _native_scheduler_followup_argv(raw_argv: list[str]) -> list[str] | None:
"""Select only generated, receipt-bound scheduler follow-up commands."""

Expand Down Expand Up @@ -92,6 +102,22 @@ def _native_scheduler_followup_argv(raw_argv: list[str]) -> list[str] | None:
/ "scheduler"
/ "heartbeat_followup_cli.ts"
)
if positionals[1] == "scheduler-ack-current" and (
"--host-match-observed" in raw_argv
and "--trae_app" not in raw_argv
):
# The bound hint predates the host update. Read the App again; only the
# typed commit owner may decide whether that observation proves an ACK.
from .upgrade import resolve_codex_app_automation_rrule

observed = resolve_codex_app_automation_rrule(
goal_id=_option_value(raw_argv, "--goal-id"),
agent_id=_option_value(raw_argv, "--agent-id"),
)
raw_argv = [
*raw_argv, "--app-automation-current-rrule",
str(observed.get("rrule") or "") if observed.get("available") else "",
]
return [
node,
"--no-warnings",
Expand Down
26 changes: 26 additions & 0 deletions tests/control_plane/test_scheduler_ack_current_host_binding.py
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,32 @@ def test_scheduler_ack_current_replays_host_binding_after_update(
assert settled_app["host_action"] == "none"


@pytest.mark.parametrize("host_missing", [False, True])
def test_scheduler_ack_current_rejects_claim_before_matching_host_readback(
tmp_path: Path, monkeypatch, host_missing: bool,
) -> None:
registry_path, runtime_root, project = write_cli_fixture(
tmp_path / "fixture", scoped_agents=True,
)
codex_home = tmp_path / "codex-home"
monkeypatch.setenv("CODEX_HOME", str(codex_home))
monkeypatch.setenv("CODEX_THREAD_ID", "fixture-thread")
_write_heartbeat_rrule(codex_home, "FREQ=MINUTELY;INTERVAL=3")
app = _quota(registry_path, runtime_root, project)["scheduler_hint"]["codex_app"]
if host_missing:
(codex_home / "automations/fixture/automation.toml").unlink()

_, ack = run_json_cli_result(
*app["ack_hint"]["cli_args"], registry_path=registry_path,
runtime_root=runtime_root, cwd=project,
)
assert ack["ok"] is False
assert ack["scheduler_commit"]["reason_code"] == "rrule_mismatch"
assert not Path(ack["scheduler_commit"]["path"]).exists()
assert ack["appended"] is False
assert ack["registry_mutated"] is False


def test_scheduler_ack_current_resets_stale_identity_from_matching_host(
tmp_path: Path,
monkeypatch,
Expand Down
22 changes: 22 additions & 0 deletions tests/control_plane_ts/scheduler_heartbeat_commit.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,25 @@ test("exact ACK commits scheduler state and is replay-safe", async (t) => {
assert.equal(replay.replayed, true);
assert.equal(replay.state_digest, first.state_digest);
});

test("host-match ACK requires a matching observation before any state write", async (t) => {
const runtimeRoot = await tempRuntime(t);
for (const observed of ["", "FREQ=MINUTELY;INTERVAL=3"]) {
const refused = await evaluateSchedulerHeartbeatCommit(request(runtimeRoot, {
host_match_observed: true,
observed_host_rrule: observed,
}));
assert.equal(refused.status, "conflict");
assert.equal(refused.reason_code, "rrule_mismatch");
assert.equal(refused.written, false);
await assert.rejects(readFile(refused.path, "utf8"), { code: "ENOENT" });
}
const accepted = await evaluateSchedulerHeartbeatCommit(request(runtimeRoot, {
host_match_observed: true,
observed_host_rrule: "FREQ=MINUTELY;INTERVAL=15",
}));
assert.equal(accepted.status, "written");
});
test("steady-state ACK is reduced to a typed skipped receipt", async (t) => {
const runtimeRoot = await tempRuntime(t);
const first = await evaluateSchedulerHeartbeatCommit(request(runtimeRoot));
Expand Down Expand Up @@ -213,6 +232,7 @@ test("CAS rejects a stale writer and leaves the newer state unchanged", async (t
expected_rrule: "FREQ=MINUTELY;INTERVAL=30",
applied_rrule: "FREQ=MINUTELY;INTERVAL=30",
host_match_observed: true,
observed_host_rrule: "FREQ=MINUTELY;INTERVAL=30",
expected_state_digest: first.state_digest,
generated_at: "2026-08-24T08:01:00Z",
}));
Expand Down Expand Up @@ -496,6 +516,7 @@ test(
apply_needed: false,
host_match_observed: true,
generated_at: "2026-08-24T08:00:00Z",
observed_host_rrule: "FREQ=MINUTELY;INTERVAL=15",
execute: true,
});

Expand Down Expand Up @@ -555,6 +576,7 @@ test("exact host-match ACK binds a fresh reset identity", async (t) => {
expected_rrule: "FREQ=MINUTELY;INTERVAL=15",
applied_rrule: "FREQ=MINUTELY;INTERVAL=15",
host_match_observed: true,
observed_host_rrule: "FREQ=MINUTELY;INTERVAL=15",
expected_state_digest: first.state_digest,
}));

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@ function request(
progression_minutes: [15, 30, 60],
expected_rrule: "FREQ=MINUTELY;INTERVAL=15",
applied_rrule: "FREQ=MINUTELY;INTERVAL=15",
observed_host_rrule: "FREQ=MINUTELY;INTERVAL=15",
cadence_class: "active_work",
generated_at: "2026-08-27T06:30:00Z",
execute: true,
Expand Down
17 changes: 17 additions & 0 deletions tests/control_plane_ts/scheduler_heartbeat_followup_cli.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,8 @@ function argv(runtimeRoot: string, format = "json"): string[] {
"-A",
"--applied-rrule",
"FREQ=MINUTELY;INTERVAL=15",
"--app-automation-current-rrule",
"FREQ=MINUTELY;INTERVAL=15",
"--host-match-observed",
];
for (const chunk of chunks(hintPayload())) {
Expand Down Expand Up @@ -198,6 +200,21 @@ test("native host-followup CLI decodes bounded argv facts and writes JSON", asyn
);
});

test("native ACK cannot use its pre-update hint as fresh host readback", async (t) => {
const runtimeRoot = await tempRuntime(t);
await writeReceipt(runtimeRoot);
for (const observed of ["", "FREQ=MINUTELY;INTERVAL=3"]) {
const args = argv(runtimeRoot);
args[args.indexOf("--app-automation-current-rrule") + 1] = observed;
const child = runCli(args);
assert.equal(child.status, 1);
const result = JSON.parse(child.stdout) as Record<string, unknown>;
assert.equal(result.ok, false);
assert.equal(result.reason_code, "rrule_mismatch");
assert.equal((result.scheduler_commit as Record<string, unknown>).written, false);
}
});

test("native host-followup CLI preserves the scheduler Markdown surface", async (t) => {
const runtimeRoot = await tempRuntime(t);
await writeReceipt(runtimeRoot);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ function followupRequest(
? {
operation,
applied_rrule: "FREQ=MINUTELY;INTERVAL=15",
observed_host_rrule: "FREQ=MINUTELY;INTERVAL=15",
source: "quota_scheduler_ack",
}
: {
Expand Down
19 changes: 19 additions & 0 deletions tests/test_cli_entrypoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -677,6 +677,25 @@ def test_console_entrypoint_selects_only_receipt_bound_native_scheduler_followup
)


@pytest.mark.parametrize("observation", [
{"available": True, "rrule": "FREQ=MINUTELY;INTERVAL=3"},
{"available": False},
])
def test_native_scheduler_reads_host_instead_of_the_applied_claim(monkeypatch, observation) -> None:
from loopx import entrypoint, upgrade

monkeypatch.setattr(shutil, "which", lambda value: "/fixture/node")
seen = []
monkeypatch.setattr(upgrade, "resolve_codex_app_automation_rrule", lambda **kw: seen.append(kw) or observation)
args = ["quota", "scheduler-ack-current", "--goal-id", "goal", "--agent-id", "agent",
"--scheduler-host-facts-chunk", "facts", "--turn-instance-id", "turn",
"--host-match-observed", "--applied-rrule", "FREQ=MINUTELY;INTERVAL=30",
"--app-automation-current-rrule", "FREQ=MINUTELY;INTERVAL=30"]
native = entrypoint._native_scheduler_followup_argv(args)
assert seen == [{"goal_id": "goal", "agent_id": "agent"}]
assert native[-2:] == ["--app-automation-current-rrule", observation.get("rrule", "")]


def test_explicit_main_argv_remains_an_in_process_compatibility_call(
monkeypatch,
) -> None:
Expand Down
Loading