Skip to content

Commit 64bfb83

Browse files
authored
Merge pull request #6325 from loopx-project/codex/monitor-unavailable-late-poll-20261013
2 parents 7177554 + 8d7b0f1 commit 64bfb83

8 files changed

Lines changed: 534 additions & 13 deletions

‎docs/reference/protocols/quota-blocked-causal-closeout-v0.md‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,21 @@ cadence and expiry remain unchanged. It creates neither a retry clock nor an
111111
advancement wait, material successor, completion or delivery claim. Existing
112112
advancement retry/causal-wait semantics and genuine committed polls retain their
113113
behavior. A committed poll cannot be rewritten as an unavailable attempt.
114+
Conversely, a committed unavailable closeout rejects a later material or
115+
non-material poll on that same exact Turn binding before provider writeback,
116+
including dry-run provider plans. Use a freshly admitted Turn for a new
117+
observation; its poll cannot change the original frozen closeout. Genuine poll
118+
replay and independently admitted auxiliary Monitors remain valid.
119+
A stored poll transaction may already have committed its provider observation
120+
while its quota ledger is still pending. Unavailable closeout rejects that
121+
competing outcome until the exact poll is recovered or its owner proves no
122+
effect. Preserve the original transaction and business receipts; do not erase
123+
them or close the attempt by claiming that its observation was unavailable.
124+
New exact-source pending receipts retain their existing GoalRef. Foreign durable
125+
or scoped pending receipts do not prove a current instance's observation. A
126+
historical pending receipt lacking GoalRef in an exact-source context requires
127+
owner reconciliation; it remains preserved. Alias-scope recovery keeps its
128+
existing compatibility.
114129

115130
Exact replay returns the frozen receipt, with no duplicate writeback or debit.
116131
After closing the original effect, reenter the current recovery host Turn to
@@ -207,6 +222,16 @@ excluded/bound 限制仍拒绝。恢复后,模式要求执行 fencing 时获
207222
检查/到期时钟、hash、material generation、cadence 与 expiry 不变。不生成重试时钟、
208223
advancement 等待、material successor、完成或投递声明。旧 advancement 重试/因果等待
209224
与真实 poll 保持原行为;已提交 poll 不能改写为不可观察尝试。
225+
反向提交顺序也互斥:不可观察结算已提交后,同一精确 Turn 身份的 material 或
226+
non-material poll 在 provider 写回前拒绝,dry-run 也不返回可写入计划。
227+
新观察须使用重新准入的新 Turn;它的 poll 不改变原冻结结算。
228+
真实 poll 的幂等重放及独立准入的 auxiliary Monitor 仍可执行。
229+
持久 poll 事务的 provider 观察可能已经提交,而 quota ledger 仍处于 pending。
230+
不可观察结算须拒绝这一竞争结果,先恢复原精确 poll 或由其 owner 证明无 effect。
231+
保留原事务及业务回执;不能删除它们或声称观察不可取得以关闭该尝试。
232+
新 exact-source pending 回执保留既有 GoalRef;其他实例的 durable 或带作用域的
233+
pending 回执不证明当前实例的观察。exact-source 环境下,历史 pending 缺 GoalRef
234+
时须核对原 owner 并保留回执;alias 作用域沿用既有恢复兼容。
210235

211236
精确重放返回冻结回执,不重复写回或扣额;原 effect 关闭后,重新进入当前 recovery
212237
host Turn 可选择独立合法任务,不新造第三身份或重绑旧 Turn。缺准入能力、缺证或错绑

‎loopx/control_plane/quota/monitor_poll_commit.ts‎

Lines changed: 69 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import { basename, dirname, join, resolve } from "node:path";
88
import { monitorSuccessorIntent, monitorSuccessorRoute } from "../scheduler/monitor_successor.ts";
99
import { normalizeTodoCapabilities } from "../todos/work_requirements.ts";
1010
import { parseProjectionDelivery } from "../todos/projection_delivery.ts";
11+
import { isMonitorUnavailableWait } from "./blocked_wait.ts";
1112
import {
1213
QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA,
1314
readQuotaSettlementForAdmittedOwnerFromSnapshot,
@@ -16,11 +17,12 @@ import {
1617
import {
1718
parseQuotaAccountingOwner,
1819
quotaGoalRef,
20+
quotaOwnerOwnsProjection,
1921
withBorrowedQuotaAccountingOwner,
2022
type QuotaAccountingOwner,
2123
} from "./source_admission.ts";
2224

23-
import type { JsonObject } from "../effect_program.ts";
25+
import type { JsonObject, SettlementIdentity } from "../effect_program.ts";
2426
import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
2527
import {
2628
appendJsonLine,
@@ -563,17 +565,21 @@ interface Admission {
563565
external: boolean;
564566
}
565567

566-
async function readAuxiliarySettlement(request: MonitorRequest): Promise<JsonObject | null> {
568+
async function readMonitorSettlement(request: MonitorRequest): Promise<JsonObject | null> {
569+
const todoId = request.observation.settlement_todo_id ?? request.observation.todo_id;
570+
// The Turn/effect and run record belong to the decision Agent; provider actor
571+
// metadata cannot select a different historical settlement to evade its fence.
572+
const agentId = request.decision.agent_id;
567573
if (!request.runtime_root || !request.turn_instance_id ||
568-
!request.observation.actor_agent_id || !request.observation.settlement_todo_id) return null;
574+
!agentId || !todoId) return null;
569575
const snapshot = await readQuotaSettlementSnapshot(
570576
request.runtime_root,
571577
request.goal_id,
572578
);
573579
return readQuotaSettlementForAdmittedOwnerFromSnapshot({
574580
schema_version: QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA,
575581
runtime_root: request.runtime_root, goal_id: request.goal_id,
576-
agent_id: request.observation.actor_agent_id, todo_id: request.observation.settlement_todo_id,
582+
agent_id: agentId, todo_id: todoId,
577583
turn_instance_id: request.turn_instance_id, replan_obligation_id: null,
578584
infer_turn_instance_id: false, allow_unbound_binding: false,
579585
}, request.owner, snapshot);
@@ -604,7 +610,7 @@ async function auxiliaryMonitorAllowed(
604610
todo.excluded_agents.some(value => typeof value !== "string") ||
605611
todo.excluded_agents.includes(decision.agent_id)))) throw conflict();
606612
if (!historicalAdmission) {
607-
const settlement = await readAuxiliarySettlement(request);
613+
const settlement = await readMonitorSettlement(request);
608614
const progress = jsonObject(settlement?.progress);
609615
// The advancement's single debit is not an observation-admission fence.
610616
// A settled exact binding still permits independently admitted due Monitors.
@@ -632,6 +638,22 @@ async function auxiliaryMonitorAllowed(
632638
}
633639

634640
async function admission(request: MonitorRequest, historicalAdmission = false): Promise<Admission> {
641+
const observation = request.observation;
642+
const settlementTodo = observation.settlement_todo_id ?? observation.todo_id;
643+
if (settlementTodo && (!observation.todo_id || settlementTodo === observation.todo_id)) {
644+
const settlement = await readMonitorSettlement(request);
645+
// The readback owner verifies admission, exact identity and durable blocked
646+
// evidence. A late observation cannot replace that frozen no-spend outcome.
647+
// Durable poll replay returns before admission; independent auxiliary
648+
// Monitors retain their advancement-bound admission below.
649+
if (jsonObject(settlement?.progress)?.closeout_kind === "typed_blocked_writeback_no_spend" &&
650+
isMonitorUnavailableWait(jsonObject(settlement?.writeback_run)?.blocked_retry, settlementTodo)) {
651+
throw new EffectRuntimeRequestError(
652+
"monitor-poll conflicts with the committed unavailable attempt; use a fresh Turn for a new observation",
653+
"monitor_poll_admission_rejected",
654+
);
655+
}
656+
}
635657
const blocked = blockedSuccessorAllowed(request.decision);
636658
const external = externalMonitorAllowed(request.decision);
637659
const auxiliary = await auxiliaryMonitorAllowed(request, historicalAdmission);
@@ -1562,7 +1584,7 @@ function successorReceipts(providerReceipt: JsonObject | null): JsonObject[] {
15621584
async function projectAuxiliaryContinuation(request: MonitorRequest, payload: JsonObject): Promise<void> {
15631585
const continuation = jsonObject(payload.turn_continuation);
15641586
if (!request.execute || continuation?.settlement_binding_matches_observation !== false) return;
1565-
const settlement = await readAuxiliarySettlement(request);
1587+
const settlement = await readMonitorSettlement(request);
15661588
if (jsonObject(settlement?.progress)?.state !== "settled") return;
15671589
// This is current readback, not a change to the historical Monitor receipt
15681590
// or permission for another advancement. Replays must not reopen a paid Turn.
@@ -1695,6 +1717,7 @@ function receiptObject(value: unknown): MonitorReceipt {
16951717
"receipt.expected_index_digest",
16961718
),
16971719
expected_index_bytes: expectedBytes,
1720+
...(receipt.goal_ref === undefined ? {} : {goal_ref: requiredObject(receipt.goal_ref, "receipt.goal_ref")}),
16981721
} as const;
16991722
const status = requireStringLiteral(
17001723
receipt.status,
@@ -1745,6 +1768,45 @@ async function readReceipt(path: string): Promise<MonitorReceipt | null> {
17451768
}
17461769
}
17471770

1771+
/** A stored observation may already have reached its provider even before the
1772+
* run index is settled. Its existing receipt owner must recover or prove no
1773+
* effect before a competing unavailable closeout can be admitted. */
1774+
export async function hasMonitorPollTransaction(
1775+
runtimeRoot: string, identity: SettlementIdentity, owner: QuotaAccountingOwner,
1776+
): Promise<boolean> {
1777+
if (!identity.todo_id) return false;
1778+
const runsDir = join(runtimeRoot, "goals", identity.goal_id, "runs");
1779+
const base = `quota-monitor-poll:${identity.goal_id}:${identity.agent_id}:${identity.turn_instance_id}`;
1780+
for (const effectId of [base, `${base}:todo:${identity.todo_id}`]) {
1781+
const receipt = await readReceipt(transactionPath(runsDir, effectId));
1782+
if (!receipt) continue;
1783+
validateReceiptScope(receipt, runsDir, join(runsDir, "index.jsonl"));
1784+
const facts = receipt.status === "provider_pending" ? receipt.provider_plan : receipt.index_record;
1785+
if (receipt.effect_id !== effectId || facts.goal_id !== identity.goal_id ||
1786+
facts.todo_id !== identity.todo_id) continue;
1787+
if (receipt.status === "provider_pending") {
1788+
// The provider plan names its actor, not the Turn's accounting Agent.
1789+
// Legacy pending receipts retain this exact accounting effect identity.
1790+
if (receipt.schema_version === MONITOR_PENDING_ADMISSION_SCHEMA &&
1791+
(receipt.admitted_decision.goal_id !== identity.goal_id ||
1792+
receipt.admitted_decision.agent_id !== identity.agent_id)) {
1793+
throw new EffectRuntimeRequestError(
1794+
"stored Monitor admission conflicts with its accounting identity", "malformed_transaction_receipt",
1795+
);
1796+
}
1797+
} else if (facts.agent_id !== identity.agent_id) continue;
1798+
const goalRef = receipt.status === "provider_pending" ? receipt.goal_ref : facts.goal_ref;
1799+
if (receipt.status === "provider_pending" && goalRef === undefined && owner.kind !== "alias") {
1800+
throw new EffectRuntimeRequestError(
1801+
"historical pending Monitor transaction lacks GoalRef scope; reconcile its original owner before unavailable closeout",
1802+
);
1803+
}
1804+
if (!quotaOwnerOwnsProjection(owner, goalRef)) continue;
1805+
return true;
1806+
}
1807+
return false;
1808+
}
1809+
17481810
function validateReceiptScope(
17491811
receipt: MonitorReceipt,
17501812
runsDir: string,
@@ -2311,6 +2373,7 @@ async function evaluateQuotaMonitorPollRequest(
23112373
expected_index_bytes: indexBytes?.length ?? 0,
23122374
provider_plan: plan,
23132375
admitted_decision: request.decision,
2376+
...(quotaGoalRef(request.owner) === null ? {} : {goal_ref: quotaGoalRef(request.owner)}),
23142377
} satisfies PendingMonitorReceipt;
23152378
await atomicWriteJson(receiptPath, pending);
23162379
}

‎loopx/control_plane/quota/settlement_readback.ts‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import {
2121
type SettlementResult,
2222
} from "../effect_program.ts";
2323
import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
24+
import { hasMonitorPollTransaction } from "./monitor_poll_commit.ts";
2425
import {
2526
jsonObject,
2627
requireJsonObject,
@@ -1373,9 +1374,21 @@ export async function readQuotaSettlement(value: unknown): Promise<JsonObject> {
13731374
: withQuotaAccountingOwner;
13741375
return await withOwner(
13751376
request.owner,
1376-
async () => readQuotaSettlementFromRequest(
1377-
request,
1378-
await readQuotaSettlementSnapshot(request.runtime_root, request.goal_id),
1379-
),
1377+
async () => {
1378+
const result = readQuotaSettlementFromRequest(request,
1379+
await readQuotaSettlementSnapshot(request.runtime_root, request.goal_id));
1380+
const identity = jsonObject(jsonObject(jsonObject(result.identity)?.result)?.value);
1381+
if (identity && request.refresh_retry &&
1382+
jsonObject(result.refresh_recovery)?.decision !== "reject" &&
1383+
isTurnScopedSettlementOutcome(request.refresh_retry.delivery_outcome,
1384+
request.refresh_retry.progress_observation, optionalString(identity.todo_id)) &&
1385+
request.refresh_retry.delivery_outcome === "outcome_gap" &&
1386+
await hasMonitorPollTransaction(request.runtime_root, identity as unknown as SettlementIdentity, request.owner)) {
1387+
throw new EffectRuntimeRequestError(
1388+
"a stored Monitor poll may already have committed its observation; recover the exact monitor-poll before an unavailable closeout",
1389+
);
1390+
}
1391+
return result;
1392+
},
13801393
);
13811394
}
Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
"""Failure-only observations for the copied runtime retirement fixture."""
2+
from __future__ import annotations
3+
4+
import json
5+
import os
6+
from pathlib import Path
7+
8+
from loopx.control_plane import effect_runtime
9+
10+
PREFIX = "loopx_fixture_retirement:"
11+
EVENTS = frozenset({"shutdown_requested", "close", "flush_enter", "flush_done", "locator_read_done",
12+
"locator_read_error", "locator_unlink_enter", "locator_unlink_done", "locator_unlink_error",
13+
"cleanup_finally", "lock_attempt", "lock_acquired", "lock_acquire_error", "lock_release_error",
14+
"lock_release_result"})
15+
CODES = frozenset({"EACCES", "EPERM", "ENOENT", "EEXIST", "EIO", "EBUSY", "ENOTEMPTY", "EBADF",
16+
"ENOSPC", "EMFILE", "ENFILE", "ETIMEDOUT", "EINTR", "ENOTDIR", "EROFS", "mutation_lock_timeout"})
17+
TRACE = '''
18+
function fixtureTrace(event: string, details: Record<string, unknown> = {}): void {
19+
try { console.error("loopx_fixture_retirement:" + JSON.stringify({
20+
event, pid: process.pid, ms: performance.now(), ...details})); } catch {}
21+
}
22+
function fixtureCode(error: unknown): string | null {
23+
const code = (error as {code?: unknown})?.code;
24+
return typeof code === "string" ? code : null;
25+
}
26+
'''
27+
28+
29+
def _replace(source: str, before: str, after: str) -> str:
30+
assert source.count(before) == 1, "copied runtime diagnostic anchor changed"
31+
return source.replace(before, after, 1)
32+
33+
34+
def instrument_copied_runtime(root: Path) -> None:
35+
"""Observe the private fixture copy; production modules are never edited."""
36+
server = root / "effect_runtime_server.ts"
37+
text = server.read_text()
38+
text = _replace(text, 'const REQUEST_SCHEMA =', TRACE + '\nconst REQUEST_SCHEMA =')
39+
for before, after in (
40+
('shutdownRequested = true;', 'shutdownRequested = true; fixtureTrace("shutdown_requested");'),
41+
('server.on("close", () => {', 'server.on("close", () => { fixtureTrace("close", {pending: pendingRequests});'),
42+
('flushCompileCache();', 'fixtureTrace("flush_enter"); flushCompileCache(); fixtureTrace("flush_done");'),
43+
('published = JSON.parse(await readFile(infoPath, "utf8"));',
44+
'published = JSON.parse(await readFile(infoPath, "utf8")); fixtureTrace("locator_read_done");'),
45+
('} catch {\n return;\n }\n // A timed-out client',
46+
'} catch (error) {\n fixtureTrace("locator_read_error", {code: fixtureCode(error)});\n return;\n }\n // A timed-out client'),
47+
('await rm(infoPath, { force: true });',
48+
'fixtureTrace("locator_unlink_enter");\n try { await rm(infoPath, { force: true }); fixtureTrace("locator_unlink_done"); }\n'
49+
' catch (error) { fixtureTrace("locator_unlink_error", {code: fixtureCode(error)}); throw error; }'),
50+
('})().finally(() => process.exit(0));',
51+
'})().finally(() => { fixtureTrace("cleanup_finally"); process.exit(0); });'),
52+
):
53+
text = _replace(text, before, after)
54+
server.write_text(text)
55+
io = root / "effect_runtime_io.ts"
56+
text = io.read_text()
57+
text = _replace(text, 'const MUTATION_LOCK_TIMEOUT_MS =', TRACE + '\nconst MUTATION_LOCK_TIMEOUT_MS =')
58+
text = _replace(text, 'const lock = await acquireFileMutationLock(targetPath, process.pid, timeoutMs);',
59+
'fixtureTrace("lock_attempt");\n let lock: FileMutationLock;\n try { lock = await acquireFileMutationLock(targetPath, process.pid, timeoutMs); }\n'
60+
' catch (error) { fixtureTrace("lock_acquire_error", {code: fixtureCode(error)}); throw error; }\n fixtureTrace("lock_acquired");')
61+
text = _replace(text, 'if (suppressErrors) return false;',
62+
'fixtureTrace("lock_release_error", {code: fixtureCode(error)});\n if (suppressErrors) return false;')
63+
text = _replace(text, 'await releaseFileMutationLock(lock.targetPath, lock.token, null, true);',
64+
'const released = await releaseFileMutationLock(lock.targetPath, lock.token, null, true);\n'
65+
' fixtureTrace("lock_release_result", {released});')
66+
io.write_text(text)
67+
68+
69+
def trace_events(raw: bytes) -> list[dict]:
70+
"""Never publish raw stderr or unknown fields, event names, or error text."""
71+
result = []
72+
for line in raw.decode("utf-8", errors="replace").splitlines():
73+
if not line.startswith(PREFIX):
74+
continue
75+
try:
76+
value = json.loads(line[len(PREFIX):])
77+
except ValueError:
78+
continue
79+
if not isinstance(value, dict) or not isinstance(value.get("event"), str) or value["event"] not in EVENTS:
80+
continue
81+
item = {"event": value["event"]}
82+
for key in ("pid", "ms", "pending"):
83+
number = value.get(key)
84+
if type(number) in (int, float):
85+
item[key] = number
86+
if type(value.get("released")) is bool:
87+
item["released"] = value["released"]
88+
if "code" in value:
89+
item["code"] = value["code"] if isinstance(value["code"], str) and value["code"] in CODES else None
90+
result.append(item)
91+
return result[-64:]
92+
93+
94+
def file_state(path: Path, pid: int) -> dict:
95+
"""Return only existence, identity matches, liveness and numeric errno."""
96+
try:
97+
value = json.loads(path.read_text())
98+
except OSError as error:
99+
return {"read": "missing" if isinstance(error, FileNotFoundError) else "error",
100+
"errno": error.errno, "winerror": getattr(error, "winerror", None)}
101+
except ValueError:
102+
return {"read": "invalid_json"}
103+
owner = value.get("pid") if isinstance(value, dict) else None
104+
return {"read": "ok", "owner_matches_runtime": owner == pid,
105+
"owner_alive": effect_runtime._pid_is_alive(owner)}
106+
107+
108+
def retained_stderr(process_args: list, options: dict, root: Path, captures: list[int]) -> None:
109+
if str(root / "effect_runtime_server.ts") in process_args and "stderr" in options:
110+
captures.append(os.dup(options["stderr"].fileno()))
111+
112+
113+
def failure_trace(captures: list[int]) -> list[dict]:
114+
events = []
115+
for descriptor in captures:
116+
try:
117+
# Observe only after the original assertion failed. dup shares the
118+
# child's file offset: seek/read/restore can race a concurrent write.
119+
# This bounded trace is evidence available now, not a complete log.
120+
position = os.lseek(descriptor, 0, os.SEEK_CUR)
121+
try:
122+
os.lseek(descriptor, 0, os.SEEK_SET)
123+
events.extend(trace_events(os.read(descriptor, 16 * 1024)))
124+
finally:
125+
os.lseek(descriptor, position, os.SEEK_SET)
126+
except OSError as error:
127+
events.append({"trace_read_errno": error.errno})
128+
return events

0 commit comments

Comments
 (0)