diff --git a/docs/architecture/rfcs/automatic-execution-admission-v0.md b/docs/architecture/rfcs/automatic-execution-admission-v0.md index 49e8654aee..852bf39b9d 100644 --- a/docs/architecture/rfcs/automatic-execution-admission-v0.md +++ b/docs/architecture/rfcs/automatic-execution-admission-v0.md @@ -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 diff --git a/examples/control_plane/quota-scheduler-registry-route-smoke.py b/examples/control_plane/quota-scheduler-registry-route-smoke.py index 3fe941ef5b..592643021d 100644 --- a/examples/control_plane/quota-scheduler-registry-route-smoke.py +++ b/examples/control_plane/quota-scheduler-registry-route-smoke.py @@ -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: @@ -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" diff --git a/examples/control_plane/quota-scheduler-state-ack-smoke.py b/examples/control_plane/quota-scheduler-state-ack-smoke.py index 8769ba2888..143dab4850 100644 --- a/examples/control_plane/quota-scheduler-state-ack-smoke.py +++ b/examples/control_plane/quota-scheduler-state-ack-smoke.py @@ -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] @@ -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 @@ -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, @@ -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, diff --git a/examples/control_plane/quota_plan_fixtures.py b/examples/control_plane/quota_plan_fixtures.py index c1245be307..86118340ca 100644 --- a/examples/control_plane/quota_plan_fixtures.py +++ b/examples/control_plane/quota_plan_fixtures.py @@ -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" diff --git a/loopx/cli_commands/quota_scheduler_followup.py b/loopx/cli_commands/quota_scheduler_followup.py index fe39744c3d..a8ebbe64a1 100644 --- a/loopx/cli_commands/quota_scheduler_followup.py +++ b/loopx/cli_commands/quota_scheduler_followup.py @@ -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, @@ -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, diff --git a/loopx/control_plane/quota/scheduler_ack.py b/loopx/control_plane/quota/scheduler_ack.py index 744c4f19a7..e4eb795b81 100644 --- a/loopx/control_plane/quota/scheduler_ack.py +++ b/loopx/control_plane/quota/scheduler_ack.py @@ -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( diff --git a/loopx/control_plane/scheduler/heartbeat_commit.ts b/loopx/control_plane/scheduler/heartbeat_commit.ts index 4d69c996c6..2017c07143 100644 --- a/loopx/control_plane/scheduler/heartbeat_commit.ts +++ b/loopx/control_plane/scheduler/heartbeat_commit.ts @@ -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, diff --git a/loopx/control_plane/scheduler/heartbeat_followup_cli.ts b/loopx/control_plane/scheduler/heartbeat_followup_cli.ts index c75e43c895..194f949d51 100644 --- a/loopx/control_plane/scheduler/heartbeat_followup_cli.ts +++ b/loopx/control_plane/scheduler/heartbeat_followup_cli.ts @@ -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); @@ -250,7 +250,7 @@ function requestFromArgs(argv: string[]): { request: Record; 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", @@ -273,6 +273,9 @@ function requestFromArgs(argv: string[]): { request: Record; 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"), diff --git a/loopx/entrypoint.py b/loopx/entrypoint.py index 4525f65924..fadb9c4d3a 100644 --- a/loopx/entrypoint.py +++ b/loopx/entrypoint.py @@ -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.""" @@ -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", diff --git a/tests/control_plane/test_scheduler_ack_current_host_binding.py b/tests/control_plane/test_scheduler_ack_current_host_binding.py index e895352150..8369947479 100644 --- a/tests/control_plane/test_scheduler_ack_current_host_binding.py +++ b/tests/control_plane/test_scheduler_ack_current_host_binding.py @@ -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, diff --git a/tests/control_plane_ts/scheduler_heartbeat_commit.test.ts b/tests/control_plane_ts/scheduler_heartbeat_commit.test.ts index 3260069431..96cc0bacd6 100644 --- a/tests/control_plane_ts/scheduler_heartbeat_commit.test.ts +++ b/tests/control_plane_ts/scheduler_heartbeat_commit.test.ts @@ -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)); @@ -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", })); @@ -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, }); @@ -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, })); diff --git a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts index d212b63b66..a636ce3df3 100644 --- a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts +++ b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts @@ -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, diff --git a/tests/control_plane_ts/scheduler_heartbeat_followup_cli.test.ts b/tests/control_plane_ts/scheduler_heartbeat_followup_cli.test.ts index f18d5ffc5a..3ba548d8d9 100644 --- a/tests/control_plane_ts/scheduler_heartbeat_followup_cli.test.ts +++ b/tests/control_plane_ts/scheduler_heartbeat_followup_cli.test.ts @@ -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())) { @@ -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; + assert.equal(result.ok, false); + assert.equal(result.reason_code, "rrule_mismatch"); + assert.equal((result.scheduler_commit as Record).written, false); + } +}); + test("native host-followup CLI preserves the scheduler Markdown surface", async (t) => { const runtimeRoot = await tempRuntime(t); await writeReceipt(runtimeRoot); diff --git a/tests/control_plane_ts/scheduler_heartbeat_receipt_fence.test.ts b/tests/control_plane_ts/scheduler_heartbeat_receipt_fence.test.ts index 24569c5a74..37d9f834f5 100644 --- a/tests/control_plane_ts/scheduler_heartbeat_receipt_fence.test.ts +++ b/tests/control_plane_ts/scheduler_heartbeat_receipt_fence.test.ts @@ -106,6 +106,7 @@ function followupRequest( ? { operation, applied_rrule: "FREQ=MINUTELY;INTERVAL=15", + observed_host_rrule: "FREQ=MINUTELY;INTERVAL=15", source: "quota_scheduler_ack", } : { diff --git a/tests/test_cli_entrypoint.py b/tests/test_cli_entrypoint.py index 975c61403f..d2346db9ed 100644 --- a/tests/test_cli_entrypoint.py +++ b/tests/test_cli_entrypoint.py @@ -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: