fix(control-plane): resume unstarted cadence reservations

docs/architecture/rfcs/automatic-execution-admission-v0.md

@@ -233,11 +233,12 @@ the interval. The local CLI remains a same-UID trust boundary.
 A managed start is two-phase in the same store: admission reserves the interval
 slot, and the Turn executor confirms that reservation only after the host
 attempt is durable in its journal. A crash between the two leaves the
-reservation resumable by the same Turn identity once the floor is reached, so a
-reserved-but-unstarted start never strands a Turn; a confirmed start stays
-fail-closed for the same identity, and an explicit manual reason cannot bypass
-that. A store record written without the phase field is read as an attempted
-start, so an older or hand-edited file fails closed rather than resuming.
+reservation immediately resumable by the same Turn identity without moving the
+original interval anchor, so a reserved-but-unstarted start never strands a
+Turn. A confirmed start stays fail-closed for the same identity, and an explicit
+manual reason cannot bypass that. A store record written without the phase field
+is read as an attempted start, so an older or hand-edited file fails closed
+rather than resuming.
 
 The M3 settings companion presents the quota-owned Goal/agent/automation policy
 through one explicit Save backed by revision-locked preview, apply and readback,

docs/architecture/rfcs/automatic-execution-admission-v0.zh-CN.md

@@ -172,9 +172,9 @@ M1 对 App 调度管理有独立价值，但不代表多宿主产品旅程完成
 本地 CLI 仍以相同 OS 用户为信任边界。
 
 managed 启动在同一 store 内分两步：准入预留间隔位，Turn executor 只在该 host 尝试
-已写入 Turn journal 之后确认该预留。两步之间进程退出时，同一 Turn 身份在满足时间
-下限后仍可恢复，因此"已预留但未启动"不会永久卡住 Turn；已确认的启动对同一身份保持
-fail-closed，显式手动理由也无法绕过。缺少阶段字段的旧记录按"已尝试启动"读取，
+已写入 Turn journal 之后确认该预留。两步之间进程退出时，同一 Turn 身份可立即恢复
+该预留，且不会移动原间隔锚点，因此"已预留但未启动"不会永久卡住 Turn。已确认的启动
+对同一身份保持 fail-closed，显式手动理由也无法绕过。缺少阶段字段的旧记录按"已尝试启动"读取，
 旧版或手工改写的文件因此 fail-closed，而不是被当作可恢复预留。
 
 M3 设置页阶段成果复用 quota 权威，一次明确的保存在内部完成修订号锁定的预览、应用与读回，

loopx/control_plane/quota/automation_cadence.ts

@@ -143,10 +143,8 @@ export async function admitAutomationStart(p: JsonObject): Promise<JsonObject> {
     }
     if (held !== undefined) {
       // Same identity, no durable host attempt: keep the original interval anchor.
-      return manual === null && current.eligible_now !== true
-        ? {...current, admitted: false, reserved: false, reason: "minimum_interval_wait"}
-        : {...current, admitted: true, reserved: true, resumed: true,
-          reason: "resumed_unstarted_reservation"};
+      return {...current, admitted: true, reserved: true, resumed: true,
+        reason: "resumed_unstarted_reservation"};
     }
     if (manual === null && current.eligible_now !== true) {
       return {...current, admitted: false, reserved: false, reason: "minimum_interval_wait"};

tests/control_plane_ts/automation_cadence.test.ts

@@ -130,8 +130,8 @@ test("an unconfirmed reservation resumes while a confirmed start fails closed",
     await manage({...base, operation: "configure", expected_revision: 0, min_interval_minutes: 60,
       owner_reference: "owner-request", execute: true});
     assert.equal((await start("turn:1", 1000)).reserved, true);
-    assert.equal((await start("turn:1", 1000 + 3_600_000 - 1)).reason, "minimum_interval_wait");
-    const resumed = await start("turn:1", 1000 + 3_600_000);
+    assert.equal((await start("turn:2", 1001)).reason, "minimum_interval_wait");
+    const resumed = await start("turn:1", 1001, {trigger_at_ms: 1000});
     assert.equal(resumed.admitted, true);
     assert.equal(resumed.resumed, true);
     assert.equal(resumed.reason, "resumed_unstarted_reservation");

tests/test_loopx_turn_executor.py

@@ -1318,11 +1318,15 @@ def die_after_reservation(identity: Mapping[str, object]) -> dict[str, object]:
     assert [
         (row["state"], row["request_id"]) for row in _cadence_starts(runtime_root)
     ] == [("reserved", f"{turn_key}:1")]
-    assert not list((runtime_root / "goals" / "fixture-goal" / "turns").glob("*.json"))
+    assert not [
+        path
+        for path in (runtime_root / "goals" / "fixture-goal" / "turns").glob("*.json")
+        if not path.name.endswith(".lock.holder.json")
+    ]
     assert calls == {"host": 0, "writeback": 0, "spend": 0, "scheduler": 0}
 
     started_at_ms = int(_cadence_starts(runtime_root)[0]["started_at_ms"])
-    monkeypatch.setattr(turn_cadence, "time", _FrozenTurnClock(started_at_ms + 120_000))
+    monkeypatch.setattr(turn_cadence, "time", _FrozenTurnClock(started_at_ms + 1))
     restart = _managed_cadence(runtime_root)
     recovered = run_loopx_turn_once(
         plan, admit_start=restart.admit, confirm_start=restart.confirm, **common
@@ -1368,7 +1372,7 @@ def die_before_attempt_record(path: Path, journal: dict[str, object]) -> None:
     assert calls == {"host": 0, "writeback": 0, "spend": 0, "scheduler": 0}
 
     started_at_ms = int(_cadence_starts(runtime_root)[0]["started_at_ms"])
-    monkeypatch.setattr(turn_cadence, "time", _FrozenTurnClock(started_at_ms + 120_000))
+    monkeypatch.setattr(turn_cadence, "time", _FrozenTurnClock(started_at_ms + 1))
     restart = _managed_cadence(runtime_root)
     recovered = run_loopx_turn_once(
         plan, admit_start=restart.admit, confirm_start=restart.confirm, **common