Skip to content

Commit 7a243e0

Browse files
huangruitenghuangruiteng
authored
Settle governed continuous-monitor proposals in the Kernel (#3511)
* feat(extensions): settle governed monitor proposals Signed-off-by: huangruiteng <huangrt01@163.com> * fix(extensions): preserve monitor recovery through settlement Signed-off-by: huangruiteng <huangrt01@163.com> --------- Signed-off-by: huangruiteng <huangrt01@163.com>
1 parent ad0afab commit 7a243e0

9 files changed

Lines changed: 1281 additions & 13 deletions

File tree

‎docs/reference/extensions.md‎

Lines changed: 47 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -418,6 +418,28 @@ An external-write operation also declares a typed `todo_contract` beside
418418
`effect_class`, containing one or more lower-snake `action_kinds` and bounded
419419
`target_key_prefixes`.
420420

421+
When a long-running provider needs LoopX to keep polling its external job, the
422+
same operation may declare a `transition_contract`. This is an authority
423+
allowlist, not a provider-owned Todo schema:
424+
425+
```json
426+
{
427+
"proposal_kinds": [
428+
"continuous_monitor_upsert",
429+
"continuous_monitor_complete"
430+
],
431+
"monitor_key_prefixes": ["example-provider:"],
432+
"monitor_action_kinds": ["poll_external_run"],
433+
"monitor_target_key_prefixes": ["external-run:"],
434+
"monitor_required_capabilities": ["network"]
435+
}
436+
```
437+
438+
The profile bounds every monitor identity, action, external target, and
439+
required capability that the provider may propose. A proposal outside those
440+
bounds fails before any LoopX state write. The provider receives no registry
441+
path and never calls Todo APIs directly.
442+
421443
1. obtain `quota should-run` admission for one exact Goal, Agent, Todo, and
422444
`turn_instance_id`; the selected open Agent Todo's `action_kind` and
423445
`target_key` must match the operation profile's `todo_contract`, so an
@@ -427,23 +449,42 @@ An external-write operation also declares a typed `todo_contract` beside
427449
idempotency key;
428450
3. call `reconcile_governed_external_capability(...)` until the provider returns
429451
a terminal `loopx_external_effect_receipt_v0`;
430-
4. supply typed writeback and spend callbacks. LoopX reuses the shared Turn
452+
4. let the LoopX Kernel materialize admitted monitor upserts through the normal
453+
Todo APIs. A `running` result may only create or retarget its bounded
454+
continuous monitor, so the recovery entry remains schedulable while the
455+
external operation or its settlement is incomplete;
456+
5. supply typed writeback and spend callbacks. LoopX reuses the shared Turn
431457
settlement driver, requires the effect receipt digest in durable writeback,
432-
and never spends quota before that writeback commits.
458+
and never spends quota before that writeback commits;
459+
6. only after the shared Turn settlement commits, let the Kernel materialize an
460+
admitted terminal monitor completion. A failed writeback or spend therefore
461+
leaves the monitor open for recovery instead of closing the only retry lane.
433462

434463
The provider may return `running`, so a service-side job can outlive the bounded
435464
provider process. Start and reconcile are separately replayable from a mode-0600
436465
journal. Exact provider revision, request digest, Goal binding, settlement
437-
identity, provider effect receipt, writeback receipt, and quota receipt remain
438-
attached to the same invocation. A crash after an external effect therefore
439-
replays with the same idempotency key and reconciles the receipt instead of
440-
starting an unrelated operation. One settlement effect id owns exactly one
466+
identity, transition proposal receipts, provider effect receipt, writeback
467+
receipt, and quota receipt remain attached to the same invocation. Each
468+
materialized proposal is checkpointed immediately. Monitor upserts belong to
469+
the pre-settlement phase; monitor completion belongs to the post-settlement
470+
phase. A crash between either Todo write and its checkpoint recovers by the
471+
proposal's stable monitor key and completion identity instead of duplicating
472+
work. A crash after an external
473+
effect replays with the same idempotency key and reconciles the receipt instead
474+
of starting an unrelated operation. One settlement effect id owns exactly one
441475
material invocation: retrying the same request replays it, while attempting a
442476
different operation or input under the same Turn receipt fails before provider
443477
dispatch. A new invocation also requires `should_run=true`; an existing journal
444478
may still be recovered with its exact typed receipt after the runnable decision
445479
has changed.
446480

481+
Transition proposals are deliberately narrower than arbitrary Goal mutation.
482+
The first version supports only continuous-monitor upsert and completion. It
483+
cannot create a general advancement Todo, change a Goal, widen capability
484+
authority, complete another Agent's work, or reopen a completed monitor. The
485+
Kernel resolves one exact monitor by its admitted binding key, applies the
486+
normal ownership and completion rules, and returns a content-bounded receipt.
487+
447488
Goal enablement alone never grants write authority. Creating or updating the
448489
binding is an explicit Goal configuration change: it uses preview/apply, but it
449490
does not create a Turn or spend Turn quota. Turn scope starts only when an

‎loopx/control_plane/effect_runtime_handlers.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -311,6 +311,7 @@ export function createEffectRuntimeHandlers(
311311
effect_id: requiredString(params.effect_id, "effect_id"),
312312
result_schema: requiredString(params.result_schema, "result_schema"),
313313
effect_class: requiredString(params.effect_class, "effect_class"),
314+
transition_contract: params.transition_contract,
314315
}),
315316
],
316317
[

‎loopx/control_plane/governed_capability.ts‎

Lines changed: 238 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,14 @@
11
import type { JsonObject } from "./effect_program.ts";
2-
import { requireNonEmptyString as requiredString } from "./runtime_decode.ts";
2+
import {
3+
assertNever,
4+
requireNonEmptyString as requiredString,
5+
requireStringLiteral,
6+
} from "./runtime_decode.ts";
37

48
export const EXTERNAL_EFFECT_RECEIPT_SCHEMA_VERSION =
59
"loopx_external_effect_receipt_v0";
10+
export const CONTINUOUS_MONITOR_PROPOSAL_SCHEMA_VERSION =
11+
"loopx_continuous_monitor_proposal_v0";
612

713
const RESULT_FIELDS = new Set([
814
"schema_version",
@@ -25,6 +31,54 @@ const EFFECT_RECEIPT_FIELDS = new Set([
2531
"evidence_digest",
2632
]);
2733

34+
const TRANSITION_CONTRACT_FIELDS = new Set([
35+
"proposal_kinds",
36+
"monitor_key_prefixes",
37+
"monitor_action_kinds",
38+
"monitor_target_key_prefixes",
39+
"monitor_required_capabilities",
40+
]);
41+
const MONITOR_UPSERT_FIELDS = new Set([
42+
"schema_version",
43+
"proposal_id",
44+
"kind",
45+
"monitor_key",
46+
"text",
47+
"action_kind",
48+
"target_key",
49+
"cadence",
50+
"next_due_at",
51+
"expires_at",
52+
"required_capabilities",
53+
]);
54+
const MONITOR_COMPLETE_FIELDS = new Set([
55+
"schema_version",
56+
"proposal_id",
57+
"kind",
58+
"monitor_key",
59+
"evidence",
60+
]);
61+
const TRANSITION_PROPOSAL_KINDS = [
62+
"continuous_monitor_upsert",
63+
"continuous_monitor_complete",
64+
] as const;
65+
type TransitionProposalKind = typeof TRANSITION_PROPOSAL_KINDS[number];
66+
type ValidatedTransitionProposal =
67+
| (JsonObject & {
68+
kind: "continuous_monitor_upsert";
69+
proposal_id: string;
70+
monitor_key: string;
71+
required_capabilities: string[];
72+
})
73+
| (JsonObject & {
74+
kind: "continuous_monitor_complete";
75+
proposal_id: string;
76+
monitor_key: string;
77+
});
78+
const PROPOSAL_ID_RE = /^[a-z][a-z0-9_.:-]{2,95}$/;
79+
const MONITOR_KEY_RE = /^[a-z][a-z0-9_.-]{0,31}:[a-z][a-z0-9_.:-]{2,95}$/;
80+
const ACTION_KIND_RE = /^[a-z][a-z0-9_]{0,63}$/;
81+
2882
function requiredObject(value: unknown, label: string): JsonObject {
2983
if (typeof value !== "object" || value === null || Array.isArray(value)) {
3084
throw new Error(`${label} must be an object`);
@@ -39,6 +93,29 @@ function boundedArray(value: unknown, label: string): unknown[] {
3993
return [...value];
4094
}
4195

96+
function boundedString(value: unknown, label: string, limit: number): string {
97+
const result = requiredString(value, label);
98+
if (result.length > limit || result.includes("\n") || result.includes("\r")) {
99+
throw new Error(`${label} must be bounded single-line text`);
100+
}
101+
return result;
102+
}
103+
104+
function boundedStringArray(
105+
value: unknown,
106+
label: string,
107+
limit: number,
108+
): string[] {
109+
const result = boundedArray(value, label);
110+
if (
111+
result.length > limit ||
112+
result.some((item) => typeof item !== "string" || item.length === 0)
113+
) {
114+
throw new Error(`${label} must contain bounded non-empty strings`);
115+
}
116+
return result as string[];
117+
}
118+
42119
function requireExactFields(
43120
value: JsonObject,
44121
expected: ReadonlySet<string>,
@@ -89,6 +166,160 @@ function externalEffectReceipt(
89166
return receipt;
90167
}
91168

169+
function validateTransitionProposals(input: {
170+
value: unknown;
171+
status: string;
172+
transition_contract: unknown;
173+
}): ValidatedTransitionProposal[] {
174+
const proposals = boundedArray(
175+
input.value,
176+
"external capability result transition_proposals",
177+
);
178+
if (proposals.length === 0) {
179+
return [];
180+
}
181+
const contract = requiredObject(
182+
input.transition_contract,
183+
"governed capability transition_contract",
184+
);
185+
requireExactFields(
186+
contract,
187+
TRANSITION_CONTRACT_FIELDS,
188+
"governed capability transition_contract",
189+
);
190+
const allowedKinds = boundedStringArray(
191+
contract.proposal_kinds,
192+
"governed capability transition_contract proposal_kinds",
193+
8,
194+
);
195+
const allowedActions = boundedStringArray(
196+
contract.monitor_action_kinds,
197+
"governed capability transition_contract monitor_action_kinds",
198+
32,
199+
);
200+
const allowedMonitorKeyPrefixes = boundedStringArray(
201+
contract.monitor_key_prefixes,
202+
"governed capability transition_contract monitor_key_prefixes",
203+
32,
204+
);
205+
const allowedTargetPrefixes = boundedStringArray(
206+
contract.monitor_target_key_prefixes,
207+
"governed capability transition_contract monitor_target_key_prefixes",
208+
32,
209+
);
210+
const allowedCapabilities = boundedStringArray(
211+
contract.monitor_required_capabilities,
212+
"governed capability transition_contract monitor_required_capabilities",
213+
32,
214+
);
215+
const seenProposalIds = new Set<string>();
216+
return proposals.map((raw, index) => {
217+
const label = `external capability transition_proposals[${index}]`;
218+
const proposal = requiredObject(raw, label);
219+
const kind: TransitionProposalKind = requireStringLiteral(
220+
proposal.kind,
221+
TRANSITION_PROPOSAL_KINDS,
222+
`${label} kind`,
223+
`${label} kind is unsupported`,
224+
);
225+
if (!allowedKinds.includes(kind)) {
226+
throw new Error(`${label} kind is not admitted by transition_contract`);
227+
}
228+
if (input.status === "running" && kind !== "continuous_monitor_upsert") {
229+
throw new Error(`${label} running result may only upsert a monitor`);
230+
}
231+
if (proposal.schema_version !== CONTINUOUS_MONITOR_PROPOSAL_SCHEMA_VERSION) {
232+
throw new Error(`${label} schema_version is invalid`);
233+
}
234+
const proposalId = boundedString(
235+
proposal.proposal_id,
236+
`${label} proposal_id`,
237+
96,
238+
);
239+
if (!PROPOSAL_ID_RE.test(proposalId) || seenProposalIds.has(proposalId)) {
240+
throw new Error(`${label} proposal_id is invalid or duplicated`);
241+
}
242+
seenProposalIds.add(proposalId);
243+
const monitorKey = boundedString(
244+
proposal.monitor_key,
245+
`${label} monitor_key`,
246+
128,
247+
);
248+
if (!MONITOR_KEY_RE.test(monitorKey)) {
249+
throw new Error(`${label} monitor_key is invalid`);
250+
}
251+
if (!allowedMonitorKeyPrefixes.some((prefix) => monitorKey.startsWith(prefix))) {
252+
throw new Error(`${label} monitor_key is not admitted`);
253+
}
254+
switch (kind) {
255+
case "continuous_monitor_upsert": {
256+
requireExactFields(proposal, MONITOR_UPSERT_FIELDS, label);
257+
const actionKind = boundedString(
258+
proposal.action_kind,
259+
`${label} action_kind`,
260+
64,
261+
);
262+
if (!ACTION_KIND_RE.test(actionKind) || !allowedActions.includes(actionKind)) {
263+
throw new Error(`${label} action_kind is not admitted`);
264+
}
265+
const targetKey = boundedString(
266+
proposal.target_key,
267+
`${label} target_key`,
268+
128,
269+
);
270+
if (!allowedTargetPrefixes.some((prefix) => targetKey.startsWith(prefix))) {
271+
throw new Error(`${label} target_key is not admitted`);
272+
}
273+
const requiredCapabilities = boundedStringArray(
274+
proposal.required_capabilities,
275+
`${label} required_capabilities`,
276+
16,
277+
);
278+
if (
279+
new Set(requiredCapabilities).size !== requiredCapabilities.length ||
280+
requiredCapabilities.some((capability) =>
281+
!ACTION_KIND_RE.test(capability) || !allowedCapabilities.includes(capability)
282+
)
283+
) {
284+
throw new Error(`${label} required_capabilities are not admitted`);
285+
}
286+
return {
287+
...proposal,
288+
kind,
289+
proposal_id: proposalId,
290+
monitor_key: monitorKey,
291+
text: boundedString(proposal.text, `${label} text`, 240),
292+
action_kind: actionKind,
293+
target_key: targetKey,
294+
cadence: boundedString(proposal.cadence, `${label} cadence`, 32),
295+
next_due_at: boundedString(
296+
proposal.next_due_at,
297+
`${label} next_due_at`,
298+
64,
299+
),
300+
expires_at: boundedString(
301+
proposal.expires_at,
302+
`${label} expires_at`,
303+
64,
304+
),
305+
required_capabilities: requiredCapabilities,
306+
};
307+
}
308+
case "continuous_monitor_complete":
309+
requireExactFields(proposal, MONITOR_COMPLETE_FIELDS, label);
310+
return {
311+
...proposal,
312+
kind,
313+
proposal_id: proposalId,
314+
monitor_key: monitorKey,
315+
evidence: boundedString(proposal.evidence, `${label} evidence`, 240),
316+
};
317+
default:
318+
return assertNever(kind, `${label} kind dispatch is incomplete`);
319+
}
320+
});
321+
}
322+
92323
export function validateGovernedCapabilityAdmission(input: {
93324
admission: unknown;
94325
todo_id: string;
@@ -160,6 +391,7 @@ export function validateGovernedCapabilityResult(input: {
160391
effect_id: string;
161392
result_schema: string;
162393
effect_class: string;
394+
transition_contract?: unknown;
163395
}): { result: JsonObject; journal_status: "running" | "ready_to_settle" } {
164396
if (input.effect_class !== "external_write") {
165397
throw new Error("governed capability execution requires external_write");
@@ -183,10 +415,14 @@ export function validateGovernedCapabilityResult(input: {
183415
"observations",
184416
"domain_state_mutations",
185417
"domain_transition_receipts",
186-
"transition_proposals",
187418
]) {
188419
result[field] = boundedArray(result[field], `external capability result ${field}`);
189420
}
421+
result.transition_proposals = validateTransitionProposals({
422+
value: result.transition_proposals,
423+
status: result.status as string,
424+
transition_contract: input.transition_contract,
425+
});
190426
result.follow_up = requiredObject(
191427
result.follow_up,
192428
"external capability result follow_up",
@@ -200,7 +436,6 @@ export function validateGovernedCapabilityResult(input: {
200436
for (const field of [
201437
"domain_state_mutations",
202438
"domain_transition_receipts",
203-
"transition_proposals",
204439
]) {
205440
if ((result[field] as unknown[]).length > 0) {
206441
throw new Error(

0 commit comments

Comments
 (0)