From 0a1cbfde9b977df2b17a8d0dc5693ab0a51f9de5 Mon Sep 17 00:00:00 2001 From: duang777 Date: Sat, 17 Oct 2026 03:27:15 +0800 Subject: [PATCH 1/4] feat(quota): seal should-run projection envelope Signed-off-by: duang777 --- .../typescript-control-plane-migration-v0.md | 14 +- ...script-control-plane-migration-v0.zh-CN.md | 14 +- .../contracts/interface-budget-contract.md | 35 +- .../contracts/projection-envelope-contract.md | 8 + .../hot-path-interface-budget-smoke.py | 16 +- loopx/control_plane/quota/should_run.py | 211 +++++++++- .../testing/cli_output_differential.py | 29 +- .../presentation/renderers/quota_markdown.py | 1 + .../test_cli_output_differential.py | 36 +- tests/test_projection_envelope.py | 360 ++++++++++++++++++ 10 files changed, 665 insertions(+), 59 deletions(-) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 42939aec8b..3c7bee59ae 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1946,12 +1946,12 @@ facade-exit condition in §4), TS gathers the facts directly and the adapter is deleted. Rollout. `status` (including `--goal-id` and projection-cache hits), -`global-summary`, `global-gates`, `global-todos` and `global-risks` now carry -the envelope. Every other `collect_status` caller receives the status envelope -in its payload but does not yet emit its own. Next, in order: `quota -should-run`, `review-packet`, and Decision Context packets. A read model added -to or migrated into TypeScript emits the envelope in the same PR; §6 makes -this a promotion gate. +`global-summary`, `global-gates`, `global-todos`, `global-risks`, and `quota +should-run` now carry the envelope. Every other `collect_status` caller receives +the status envelope in its payload but does not yet emit its own. Next, in +order: `review-packet` and Decision Context packets. A read model added to or +migrated into TypeScript emits the envelope in the same PR; §6 makes this a +promotion gate. Consumers treat a missing envelope as unknown freshness, and disclose an alerting one before stating any conclusion that depends on it. Field @@ -2023,7 +2023,7 @@ choice is now implemented rather than hypothetical. | Quota monitor-poll commit transaction | TypeScript owns monitor admission revalidation, target/event/result construction, effect replay/index CAS, provider intent, and repairable JSON/Markdown/index persistence | Python projects compact `should-run` facts, invokes the real Todo provider between at most two reductions, reloads legacy status, and holds the cross-writer index lock | | Runtime decoders ([#3443](https://github.com/loopx-project/loopx/pull/3443)) | Stable primitive decoding has one small shared module; domain decoders remain local | No larger schema framework is justified | | Transaction payoff ([#3464](https://github.com/loopx-project/loopx/pull/3464), [#3481](https://github.com/loopx-project/loopx/pull/3481), and Todo completion) | Turn settlement, quota delivery routing, and Todo completion each cross one coarse TS boundary; the Todo transaction owns identity, replay fencing, validation planning/result reduction, continuation/recovery, and completion metadata | Python still executes explicitly external providers and materializes legacy Markdown results; other domains still need their own bounded cutovers | -| Projection envelope | TypeScript owns decoding of `loopx_projection_envelope_v0` and every freshness, alert, completeness and replay decision | Python gathers read facts for `status`, `global-summary`, `global-gates`, `global-todos` and `global-risks` until those projections migrate | +| Projection envelope | TypeScript owns decoding of `loopx_projection_envelope_v0` and every freshness, alert, completeness and replay decision | Python gathers read facts for `status`, `global-summary`, `global-gates`, `global-todos`, `global-risks`, and quota `should-run` until those projections migrate | | Promoted-authority Todo claim | TypeScript owns the provider-head read, lifecycle validation, complete-record update, hard-lease check, CAS, receipt, and readback-safe result for claims after authority promotion | Default local Markdown mode remains on the legacy writer; other Todo mutations and Markdown regeneration remain bounded follow-ups | The scheduler facade exit now includes its first bounded Stage 3 route. A diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 8f8555a477..31ca493d43 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -1329,12 +1329,12 @@ Python 拥有的投影只传入紧凑的读取 facts。每个投影一次请求 status projection 迁入 kernel（§4 已将其列为 facade 退出条件），由 TS 直接收集 facts，adapter 随之删除。 -推广顺序。`status`（含 `--goal-id` 与 projection cache 命中）、`global-summary` -与 `global-gates` 现已携带 envelope。其他 `collect_status` 调用方会在 payload 里 -收到 status envelope，但尚未输出自己的 envelope。下一步依次为：`global-todos` 与 -`global-risks`（同样的组合，各一次调用）、`quota should-run`、`review-packet`， -以及 Decision Context packet。新加入或迁入 TypeScript 的读模型在同一个 PR 里输出 -envelope；§6 把这一条定为 promotion 门禁。 +推广顺序。`status`（含 `--goal-id` 与 projection cache 命中）、`global-summary`、 +`global-gates`、`global-todos`、`global-risks` 与 `quota should-run` 现已携带 +envelope。其他 `collect_status` 调用方会在 payload 里收到 status envelope，但尚未 +输出自己的 envelope。下一步依次为：`review-packet` 与 Decision Context packet。 +新加入或迁入 TypeScript 的读模型在同一个 PR 里输出 envelope；§6 把这一条定为 +promotion 门禁。 消费者把缺失 envelope 视为新鲜度未知；envelope 告警时，必须先披露，再陈述依赖它的 结论。字段语义与消费者规则见 @@ -1392,7 +1392,7 @@ replay、receipt 与 settlement。这个架构选择已经落地，不再是假 | Quota monitor-poll commit transaction | TypeScript 拥有 monitor admission 复核、target/event/result 构造、effect replay/index CAS、provider intent，以及可修复的 JSON/Markdown/index persistence | Python 投影 compact `should-run` facts，在最多两次 reduction 之间调用真实 Todo provider，刷新 legacy status，并持有 cross-writer index lock | | Runtime decoder（[#3443](https://github.com/loopx-project/loopx/pull/3443)） | 稳定 primitive decoding 进入一个很小的共享模块；domain decoder 仍留在本地 | 没有理由建设更大的 schema framework | | Transaction 兑现（[#3464](https://github.com/loopx-project/loopx/pull/3464)、[#3481](https://github.com/loopx-project/loopx/pull/3481) 与 Todo completion） | Turn settlement、quota delivery routing 与 Todo completion 均只跨一个粗粒度 TS boundary；Todo transaction 拥有 identity、replay fence、validation planning/result reduction、continuation/recovery 与 completion metadata | Python 仍执行显式 external provider，并物化 legacy Markdown result；其他 domain 仍需各自的 bounded cutover | -| Projection envelope | TypeScript 拥有 `loopx_projection_envelope_v0` 的解码，以及全部 freshness、告警、完整性与重放判定 | 在 `status`、`global-summary`、`global-gates` 迁移前，Python 仍为它们收集读取 facts | +| Projection envelope | TypeScript 拥有 `loopx_projection_envelope_v0` 的解码，以及全部 freshness、告警、完整性与重放判定 | 在这些投影迁移前，Python 仍为 `status`、`global-summary`、`global-gates`、`global-todos`、`global-risks` 与 quota `should-run` 收集读取 facts | Scheduler facade exit 已交付第一段有边界的 Stage 3 路径。带版本的 `heartbeat_followup_cli.ts` 从生成的 ACK/failure hint 接收有大小上限的 compact host diff --git a/docs/reference/contracts/interface-budget-contract.md b/docs/reference/contracts/interface-budget-contract.md index d29073c8a4..193c83bf43 100644 --- a/docs/reference/contracts/interface-budget-contract.md +++ b/docs/reference/contracts/interface-budget-contract.md @@ -10,7 +10,7 @@ and size/count budgets. | --- | --- | --- | --- | --- | --- | --- | | `heartbeat_prompt_json` | heartbeat automation | wake and route one bounded turn | `quota should-run`, `status`, or `review-packet --handoff-only` | `json_chars <= 5400` plus `interface_budget.within_budget=true` | `nested_keys <= 40` | `top_level_keys <= 30` | | `review_packet_handoff_only_json` | project-agent handoff | forward the smallest sufficient task packet | full `review-packet` or run-history artifact | `json_chars <= 3000` plus `handoff_interface_budget.within_budget=true` | `nested_keys <= 40` | `top_level_keys <= 18` | -| `quota_should_run_json` | quota guard | decide whether the selected goal may spend compute | `status`, `history`, or active state | `json_chars <= 16500` | `nested_keys <= 386` | `top_level_keys <= 54` | +| `quota_should_run_json` | quota guard | decide whether the selected goal may spend compute | `status`, `history`, or active state | `json_chars <= 20500` | `nested_keys <= 525` | `top_level_keys <= 54` | | `dashboard_status_json` | operator dashboard | render first-screen operator state | `history`, run artifacts, or project-local adapter output | `json_chars <= 24000` | `nested_keys <= 350` | `top_level_keys <= 27` | These four budgets measure compact machine payloads. For @@ -23,21 +23,24 @@ indentation and Markdown wrappers can make emitted output materially larger; the emitted-output qualification matrix below measures that separate boundary through the real CLI entry point. -The current quota fixture retains 378 nested keys: the M3 automatic-cadence -readback established 368, and full-source frontier counts add ten keys for -deferred readiness and explicit known/unknown coverage. On the same synthetic -workload, base/head measure 14,897/15,115 JSON characters, 368/378 nested keys -and 52/52 top-level keys. Dashboard status grows 22,256/22,362 characters while -its 336 nested and 26 top-level keys stay unchanged. The quota nested-key -ceiling moves from 376 to 386, preserving eight keys of headroom. Character, -top-level, per-Todo growth and other surface ceilings remain unchanged. These -facts prevent hidden rows from becoming blocked work; removing the coverage -fields to meet the old ceiling would lose useful meaning. - -同一合成 workload 的 quota 在 base/head 分别为 14,897/15,115 字符、368/378 -嵌套键和 52/52 顶层键。新增十个键保留 deferred 就绪计数及 known/unknown 证据； -只将嵌套键预算从 376 调至 386，仍留八键余量。字符、顶层键、每任务增长和其他 -surface 上限不变；不通过删除来源完整性来把未展示任务误报为 blocked。 +The current quota fixture measures 15,419 characters, 378 nested keys, and 52 +top-level keys on `main`. The `quota_should_run` projection envelope raises the +same workload to 18,636 characters, 513 nested keys, and 53 top-level keys. The +new keys carry the inherited status sources, direct live-read facts, freshness, +alerts, and Goal coverage that a consumer must inspect before spending compute. +The character ceiling moves from 16,500 to 20,500, and the nested-key ceiling +moves from 386 to 525. The top-level ceiling stays at 54. The new ceilings leave +1,864 characters, 12 nested keys, and one top-level key on the measured head. +Removing envelope rows would violate the projection contract, so the budget +records the useful contract growth instead of hiding it through compaction. + +同一合成 workload 在 `main` 上为 15,419 字符、378 个嵌套键和 52 个顶层键；加入 +`quota_should_run` projection envelope 后为 18,636 字符、513 个嵌套键和 53 个 +顶层键。新增字段携带继承的 status 来源、实时读取事实、freshness、告警与 Goal +覆盖，消费者必须在花费 compute 前读取这些信息。字符上限从 16,500 调至 20,500， +嵌套键上限从 386 调至 525，顶层键上限仍为 54。精确 head 仍保留 1,864 字符、 +12 个嵌套键与一个顶层键余量。删除 envelope 行会破坏投影合同，因此这里记录有用的 +合同增长，而不是用压缩隐藏它。 The successful thin heartbeat projection contains only `schema_version`, `ok`, `goal_id`, optional `agent_id`, `task_body`, and the compact diff --git a/docs/reference/contracts/projection-envelope-contract.md b/docs/reference/contracts/projection-envelope-contract.md index 80799f6848..2b9e6c62eb 100644 --- a/docs/reference/contracts/projection-envelope-contract.md +++ b/docs/reference/contracts/projection-envelope-contract.md @@ -23,10 +23,18 @@ Python-owned projections do so through `loopx/control_plane/projection_envelope_ | `loopx global-gates` | `global_gates` | `global` | | `loopx global-todos` | `global_todos` | `global` | | `loopx global-risks` | `global_risks` | `global` | +| `loopx quota should-run` | `quota_should_run` | `goal` | Other read models adopt the envelope in the order listed in the RFC; until then they carry no freshness guarantee and consumer rule 1 below applies. +`quota_should_run` inherits the status envelope. It also records each live +source that the decision reads, including automation cadence, an applicable +App scheduler state, the selected delivery workspace, the operator inbox, and +replan exploration. A missing persisted App scheduler state is a successful +empty read. `owner_read_failed` makes automation cadence unreadable, while +`runtime_root_unavailable` means that source was not read. + ## Shape ```json diff --git a/examples/control_plane/hot-path-interface-budget-smoke.py b/examples/control_plane/hot-path-interface-budget-smoke.py index c1e1a424c3..aa2d31350c 100644 --- a/examples/control_plane/hot-path-interface-budget-smoke.py +++ b/examples/control_plane/hot-path-interface-budget-smoke.py @@ -73,15 +73,13 @@ # provider-neutral app_automation packet becomes canonical. # Full Goal/selected-work read commands add necessary source continuity. # One Agent-channel carrier avoids duplicating them in the CLI channel. - # M3 adds one validated cadence readback envelope to the quota packet: - # 357 -> 368 nested keys and 51 -> 52 top-level keys. - # Identical raw packet: 15,201 at the ordinary temp root, 15,865 at - # the original longer root. Full command routes are useful; retain - # them with 635 chars of headroom, without widening structural guards. - "max_json_chars": 16_500, - # Full-source frontier readiness/coverage adds ten useful keys: - # base 368 -> 378, preserving eight keys of structural headroom. - "max_nested_keys": 386, + # The projection envelope adds the source freshness and goal coverage + # needed to trust this decision. On the same fixture, main/head measure + # 15,419/18,636 characters, 378/513 nested keys, and 52/53 top-level + # keys. Keep the complete typed envelope rather than dropping source + # facts that the consumer must inspect before it spends compute. + "max_json_chars": 20_500, + "max_nested_keys": 525, "max_top_level_keys": 54, }, "dashboard_status_json": { diff --git a/loopx/control_plane/quota/should_run.py b/loopx/control_plane/quota/should_run.py index c9f9cc90b0..cfc4db36b1 100644 --- a/loopx/control_plane/quota/should_run.py +++ b/loopx/control_plane/quota/should_run.py @@ -29,6 +29,8 @@ registry_goal_by_id as _registry_goal_by_id, ) from ..quota.projection_repair import build_boundary_projection_repair_hint +from ..projection_envelope_facts import seal_projection_envelope, source_fact +from ..runtime.time import now_utc_iso from ..quota.selected_todo_projection import selected_todo_projection from ..quota.states import ( AutomaticTurnPauseCause, @@ -68,11 +70,155 @@ GOAL_STOPPED_MODE = "goal_stopped" +def _tracked_read_callback( + callback: Callable[..., Any] | None, + *, + source_id: str, + direct_sources: dict[str, dict[str, Any]], +) -> Callable[..., Any] | None: + """Record a supplied live reader only when quota invokes it.""" + + if callback is None: + return None + + def tracked(*args: Any, **kwargs: Any) -> Any: + read_at = now_utc_iso() + try: + result = callback(*args, **kwargs) + except (OSError, RuntimeError, ValueError): + direct_sources[source_id] = source_fact( + source_id, + read_status="unreadable", + last_read_at=read_at, + unreadable_count=1, + ) + raise + read_status = ( + "unreadable" + if isinstance(result, Mapping) + and result.get("projection_status") == "unavailable" + else "read" + ) + direct_sources[source_id] = source_fact( + source_id, + read_status=read_status, + last_read_at=read_at, + item_count=0 if result is None else 1, + unreadable_count=1 if read_status == "unreadable" else 0, + ) + return result + + return tracked + + +def _automation_cadence_source( + payload: Mapping[str, Any], *, observed_at: str +) -> dict[str, Any] | None: + readback = payload.get("automation_cadence_readback") + if not isinstance(readback, Mapping): + return None + eligibility = readback.get("eligibility") + reason = ( + str(eligibility.get("reason") or "") + if isinstance(eligibility, Mapping) + else "owner_read_failed" + ) + if reason == "runtime_root_unavailable": + return source_fact("automation_cadence", read_status="not_read") + if reason == "owner_read_failed": + return source_fact( + "automation_cadence", + read_status="unreadable", + last_read_at=observed_at, + unreadable_count=1, + ) + return source_fact( + "automation_cadence", + last_read_at=observed_at, + item_count=1, + ) + + +def _attach_quota_should_run_envelope( + payload: dict[str, Any], + status_payload: Mapping[str, Any], + *, + safe_goal_id: str, + direct_sources: Mapping[str, Mapping[str, Any]], +) -> dict[str, Any]: + observed_at = now_utc_iso() + status_envelope = status_payload.get("projection_envelope") + has_status_envelope = isinstance(status_envelope, Mapping) + cadence_source = _automation_cadence_source(payload, observed_at=observed_at) + goal_found = not ( + payload.get("state") == "unknown" + and payload.get("status") == "goal_not_found" + ) + omitted = [] + if not goal_found: + omitted.append( + { + "reason": "goal_not_found", + "count": 1, + "refs": [safe_goal_id] if safe_goal_id else [], + } + ) + payload["projection_envelope"] = seal_projection_envelope( + projection="quota_should_run", + observed_at=observed_at, + sources=[ + *[dict(row) for row in direct_sources.values()], + *([cadence_source] if cadence_source else []), + *( + [] + if has_status_envelope + else [source_fact("status", read_status="not_read")] + ), + ], + coverage={ + "scope": "goal", + "expected_count": 1, + "included_count": 1 if goal_found else 0, + "omitted": omitted, + }, + upstream=[dict(status_envelope)] if has_status_envelope else [], + ) + return payload + + +def _record_app_scheduler_state_source( + payload: Mapping[str, Any], + status_payload: Mapping[str, Any], + resolved_scheduler_context: SchedulerExecutionContextResolution, + direct_sources: dict[str, dict[str, Any]], +) -> None: + context = resolved_scheduler_context.context + if not ( + resolved_scheduler_context.ok + and context is not None + and context.app_automation_applicable + ): + return + identity = payload.get("agent_identity") + agent_id = ( + str(identity.get("agent_id") or "").strip() + if isinstance(identity, Mapping) + else "" + ) + read = bool(status_payload.get("runtime_root") and agent_id) + direct_sources["app_scheduler_state"] = source_fact( + "app_scheduler_state", + read_status="read" if read else "not_read", + last_read_at=now_utc_iso() if read else None, + ) + + def _apply_selected_todo_guards( prepared: _QuotaDecisionPreparation, route: _QuotaDecisionRoute, *, workspace_path: Path | None = None, + direct_sources: dict[str, dict[str, Any]] | None = None, ) -> _QuotaDecisionRoute: """Bind workspace and boundary guards to the exact projected Todo. @@ -136,6 +282,12 @@ def _apply_selected_todo_guards( local_workspace = observe_goal_local_workspace( prepared.item, selected_todo, goal_boundary.get("write_scope", []) ) + if direct_sources is not None: + direct_sources["delivery_workspace"] = source_fact( + "delivery_workspace", + last_read_at=now_utc_iso(), + item_count=1 if selected_todo else 0, + ) if not prepared.inbox_priority_due: workspace_guard = build_agent_workspace_guard( prepared.item, @@ -336,6 +488,17 @@ def build_quota_should_run( replan_explore_reader: Callable[[str, str | None], Mapping[str, Any] | None] | None = None, ) -> dict[str, Any]: safe_goal_id = str(goal_id or "").strip() + direct_sources: dict[str, dict[str, Any]] = {} + tracked_operator_inbox_reader = _tracked_read_callback( + operator_inbox_urgency_projector, + source_id="operator_inbox", + direct_sources=direct_sources, + ) + tracked_replan_explore_reader = _tracked_read_callback( + replan_explore_reader, + source_id="replan_explore", + direct_sources=direct_sources, + ) resolved_scheduler_context = resolve_scheduler_execution_context( scheduler_execution_context ) @@ -369,7 +532,7 @@ def build_quota_should_run( ) if item: if _quota_item_is_paused(item): - return build_quota_paused_should_run_payload( + payload = build_quota_paused_should_run_payload( status_payload, safe_goal_id=safe_goal_id, requested_agent_id=agent_id, @@ -383,6 +546,12 @@ def build_quota_should_run( runtime_root=runtime_root, goal_ref=goal_ref, ) + return _attach_quota_should_run_envelope( + payload, + status_payload, + safe_goal_id=safe_goal_id, + direct_sources=direct_sources, + ) prepared = _prepare_quota_should_run_item( status_payload, safe_goal_id=safe_goal_id, @@ -392,7 +561,7 @@ def build_quota_should_run( codex_app_current_rrule=codex_app_current_rrule, codex_app_automation_id=codex_app_automation_id, resolved_scheduler_context=resolved_scheduler_context, - operator_inbox_urgency_projector=operator_inbox_urgency_projector, + operator_inbox_urgency_projector=tracked_operator_inbox_reader, registry_goal=registry_goal, plan=plan, goal_health_ok=goal_health_ok, @@ -405,12 +574,17 @@ def build_quota_should_run( receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id, receipt_bound_replan_guard_scoped=receipt_bound_replan_guard_scoped, authoritative_fallback_todo_items=authoritative_fallback_todo_items, - replan_explore_reader=replan_explore_reader, + replan_explore_reader=tracked_replan_explore_reader, receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at, ) route = _resolve_quota_should_run_route(prepared) - route = _apply_selected_todo_guards(prepared, route, workspace_path=workspace_path) - return _build_quota_should_run_payload( + route = _apply_selected_todo_guards( + prepared, + route, + workspace_path=workspace_path, + direct_sources=direct_sources, + ) + payload = _build_quota_should_run_payload( prepared, route, turn_instance_id=turn_instance_id, @@ -418,6 +592,18 @@ def build_quota_should_run( runtime_root=runtime_root, goal_ref=goal_ref, ) + _record_app_scheduler_state_source( + payload, + status_payload, + resolved_scheduler_context, + direct_sources, + ) + return _attach_quota_should_run_envelope( + payload, + status_payload, + safe_goal_id=safe_goal_id, + direct_sources=direct_sources, + ) if health_item: payload = { "ok": False, @@ -444,8 +630,13 @@ def build_quota_should_run( runtime_root=runtime_root or status_payload.get("runtime_root"), resolved_scheduler_context=resolved_scheduler_context, ) - return payload - return { + return _attach_quota_should_run_envelope( + payload, + status_payload, + safe_goal_id=safe_goal_id, + direct_sources=direct_sources, + ) + payload = { "ok": False, "mode": "should-run", "goal_id": safe_goal_id, @@ -461,3 +652,9 @@ def build_quota_should_run( ), "plan_summary": plan.get("summary"), } + return _attach_quota_should_run_envelope( + payload, + status_payload, + safe_goal_id=safe_goal_id, + direct_sources=direct_sources, + ) diff --git a/loopx/control_plane/testing/cli_output_differential.py b/loopx/control_plane/testing/cli_output_differential.py index aac1d1d6a4..e1a2ce9914 100644 --- a/loopx/control_plane/testing/cli_output_differential.py +++ b/loopx/control_plane/testing/cli_output_differential.py @@ -671,28 +671,43 @@ def _schema_migration_growth_allowance( # The unchanged status matrix adds 2,801 pretty JSON chars / 102 lines / -# 1,910 compact chars for five source rows. This reviewed one-time transition -# leaves headroom without changing absolute ceilings or ordinary v0-to-v0 growth. +# 1,910 compact chars for five source rows. Quota should-run adds 2,617 JSON +# chars or 167 Markdown chars for its inherited and direct source rows. This +# reviewed one-time transition leaves headroom without changing absolute +# ceilings or ordinary v0-to-v0 growth. _PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE = GrowthAllowance( ratio=0, json={"chars": 3_000, "utf8_bytes": 3_000, "lines": 110, "compact_payload_chars": 2_048}, markdown={"chars": 192, "utf8_bytes": 224, "lines": 3, "compact_payload_chars": 0}, ) +_QUOTA_PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE = GrowthAllowance( + ratio=0, + json={"chars": 3_000, "utf8_bytes": 3_000, "lines": 110, "compact_payload_chars": 2_800}, + markdown={"chars": 192, "utf8_bytes": 224, "lines": 3, "compact_payload_chars": 0}, +) def _projection_envelope_migration( base: dict[str, Any], candidate: dict[str, Any], *, output_format: str, ) -> tuple[dict[Metric, int], list[str], list[str]]: - """Qualify the one-time status schema transition, never unrelated growth.""" + """Qualify the one-time projection schema transition, never unrelated growth.""" before = tuple(base.get("projection_envelope_schema_versions") or []) after = tuple(candidate.get("projection_envelope_schema_versions") or []) if before == after: return {}, [], [] - row_id = str(base["row_id"]) - if (row_id.startswith(("surface/status/", "variant/status_task_graph_detail/")) + surface_id = str(base.get("surface_id") or "") + if (surface_id in {"status", "quota_should_run"} and before == () and after == ("loopx_projection_envelope_v0",)): - allowance = (_PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE.json - if output_format == "json" else _PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE.markdown) + migration_allowance = ( + _QUOTA_PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE + if surface_id == "quota_should_run" + else _PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE + ) + allowance = ( + migration_allowance.json + if output_format == "json" + else migration_allowance.markdown + ) return allowance, [], ["projection envelope schema migrated: none -> loopx_projection_envelope_v0"] return {}, ["projection envelope schema coverage changed"], [] diff --git a/loopx/presentation/renderers/quota_markdown.py b/loopx/presentation/renderers/quota_markdown.py index 767183275d..1579807a43 100644 --- a/loopx/presentation/renderers/quota_markdown.py +++ b/loopx/presentation/renderers/quota_markdown.py @@ -352,6 +352,7 @@ def render_quota_should_run_markdown(payload: dict[str, Any]) -> str: lines = [ "# LoopX Quota Should Run", "", + *render_projection_envelope_markdown(payload.get("projection_envelope")), f"- ok: `{payload.get('ok')}`", f"- goal_id: `{payload.get('goal_id')}`", f"- decision: `{payload.get('decision')}`", diff --git a/tests/control_plane/test_cli_output_differential.py b/tests/control_plane/test_cli_output_differential.py index 68e55f3d8f..b3377de2f6 100644 --- a/tests/control_plane/test_cli_output_differential.py +++ b/tests/control_plane/test_cli_output_differential.py @@ -1279,15 +1279,31 @@ def test_measurement_only_probe_skips_ceiling_but_keeps_semantic_shape() -> None @pytest.mark.parametrize("output_format", ["json", "markdown"]) -def test_projection_envelope_migration_is_status_only_bounded_and_one_time(output_format): +@pytest.mark.parametrize( + ("surface_id", "row_prefix"), + [ + ("status", "surface/status"), + ("status", "variant/status_task_graph_detail"), + ("quota_should_run", "surface/quota_should_run"), + ("quota_should_run", "variant/quota_should_run_all_detail"), + ], +) +def test_projection_envelope_migration_is_bounded_and_one_time( + output_format, surface_id, row_prefix, +): from loopx.control_plane.testing.cli_output_semantics import projection_envelope_schema_versions assert projection_envelope_schema_versions({"projection_envelope": { "schema_version": "loopx_projection_envelope_v0"}}) == ["loopx_projection_envelope_v0"] assert projection_envelope_schema_versions("- projection: envelope=`loopx_projection_envelope_v0` observed_at=`today`") == ["loopx_projection_envelope_v0"] - limits = ({"chars": 3000, "utf8_bytes": 3000, "lines": 110, "compact_payload_chars": 2048} + compact_limit = 2800 if surface_id == "quota_should_run" else 2048 + limits = ({"chars": 3000, "utf8_bytes": 3000, "lines": 110, "compact_payload_chars": compact_limit} if output_format == "json" else {"chars": 192, "utf8_bytes": 224, "lines": 3, "compact_payload_chars": 0}) - base = _row(format=output_format, row_id=f"surface/status/small/{output_format}") + base = _row( + format=output_format, + row_id=f"{row_prefix}/small/{output_format}", + surface_id=surface_id, + ) candidate = {**base, "projection_envelope_schema_versions": ["loopx_projection_envelope_v0"], **{metric: base[metric] + limit for metric, limit in limits.items()}} result = compare_cli_output_receipts(_receipt(base), _receipt(candidate)) @@ -1305,9 +1321,17 @@ def test_projection_envelope_migration_is_status_only_bounded_and_one_time(outpu for versions in ([], ["loopx_projection_envelope_v1"]): unknown = {**candidate, "projection_envelope_schema_versions": versions} assert not compare_cli_output_receipts(_receipt(candidate), _receipt(unknown))["ok"] - for surface in ("quota_should_run", "todo_list", "status_unrelated"): - outside_base = {**base, "row_id": f"surface/{surface}/small/{output_format}"} - outside = {**candidate, "row_id": outside_base["row_id"]} + for surface in ("todo_list", "status_unrelated", "quota_should_run_unrelated"): + outside_base = { + **base, + "row_id": f"surface/{surface}/small/{output_format}", + "surface_id": surface, + } + outside = { + **candidate, + "row_id": outside_base["row_id"], + "surface_id": surface, + } assert not compare_cli_output_receipts(_receipt(outside_base), _receipt(outside))["ok"] diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py index 62a309f987..189fb578c6 100644 --- a/tests/test_projection_envelope.py +++ b/tests/test_projection_envelope.py @@ -7,9 +7,15 @@ import pytest import loopx.control_plane.projection_envelope_facts as projection_envelope +import loopx.control_plane.quota.should_run as quota_should_run_module +import loopx.control_plane.quota.should_run_packet as should_run_packet import loopx.global_risks as global_risks import loopx.global_todos as global_todos import loopx.summary_all as summary_all +from loopx.control_plane.testing.quota_fixtures import ( + quota_status_payload, + quota_todo_item, +) from loopx.control_plane.runtime.status_projection_cache import ( load_status_projection_cache, status_projection_cache_path, @@ -17,7 +23,9 @@ write_status_projection_cache, ) from loopx.control_plane.runtime.time import now_utc, utc_isoformat +from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown from loopx.presentation.renderers.status_markdown import render_status_markdown +from loopx.quota import build_quota_should_run from loopx.status import collect_status @@ -106,6 +114,358 @@ def test_goal_status_scopes_coverage_to_the_goal(tmp_path: Path) -> None: assert envelope["complete"] is True +def test_quota_should_run_inherits_status_and_covers_the_selected_goal( + tmp_path: Path, +) -> None: + status = _status(_registry(tmp_path, ("goal-a",)), goal_id="goal-a") + + payload = build_quota_should_run( + status, + goal_id="goal-a", + runtime_root=status["runtime_root"], + ) + + envelope = payload["projection_envelope"] + sources = _sources(envelope) + assert envelope["projection"] == "quota_should_run" + assert envelope["coverage"]["scope"] == "goal" + assert envelope["coverage"]["expected_count"] == 1 + assert envelope["coverage"]["included_count"] == 1 + assert envelope["coverage"]["omitted"] == [] + assert envelope["coverage"]["complete"] is True + assert envelope["complete"] is True and envelope["alert"] is False + assert envelope["upstream"][0]["projection"] == "status" + assert sources["automation_cadence"]["status"] == "fresh" + assert str(tmp_path) not in json.dumps(envelope) + + +def test_quota_should_run_alerts_when_the_status_envelope_is_missing( + tmp_path: Path, +) -> None: + status = _status(_registry(tmp_path, ("goal-a",)), goal_id="goal-a") + status.pop("projection_envelope") + + payload = build_quota_should_run( + status, + goal_id="goal-a", + runtime_root=status["runtime_root"], + ) + + envelope = payload["projection_envelope"] + assert _sources(envelope)["status"]["status"] == "not_read" + assert envelope["alert_reasons"] == ["missing_required_sources"] + markdown = render_quota_should_run_markdown(payload) + assert markdown.index("projection alerts") < markdown.index("decision:") + + +def test_quota_should_run_marks_an_unknown_goal_as_incomplete(tmp_path: Path) -> None: + status = _status(_registry(tmp_path, ("goal-a",))) + + envelope = build_quota_should_run( + status, + goal_id="goal-missing", + runtime_root=status["runtime_root"], + )["projection_envelope"] + + assert envelope["coverage"]["scope"] == "goal" + assert envelope["coverage"]["expected_count"] == 1 + assert envelope["coverage"]["included_count"] == 0 + assert envelope["coverage"]["omitted"] == [ + {"reason": "goal_not_found", "count": 1, "refs": ["goal-missing"]} + ] + assert envelope["complete"] is False + assert envelope["alert_reasons"] == ["incomplete_coverage"] + + +def test_quota_should_run_propagates_unreadable_and_stale_status_sources( + tmp_path: Path, +) -> None: + observed_at = projection_envelope.now_utc_iso() + status = quota_status_payload( + goal_id="goal-a", + status="active", + recommended_action="Advance the goal.", + ) + status["runtime_root"] = str(tmp_path / "runtime") + status["projection_envelope"] = projection_envelope.seal_projection_envelope( + projection="status", + observed_at=observed_at, + sources=[ + projection_envelope.source_fact( + "registry", + read_status="unreadable", + last_read_at=observed_at, + ), + projection_envelope.source_fact( + "goal_state_contract", + last_read_at="2000-01-01T00:00:00Z", + ), + ], + coverage={ + "scope": "goal", + "expected_count": 1, + "included_count": 1, + "omitted": [], + }, + ) + + envelope = build_quota_should_run( + status, + goal_id="goal-a", + runtime_root=status["runtime_root"], + )["projection_envelope"] + + assert "unreadable_sources" in envelope["alert_reasons"] + assert "stale_sources" in envelope["alert_reasons"] + assert envelope["complete"] is True + + +def test_quota_should_run_seals_paused_and_health_blocked_results( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + status = quota_status_payload( + goal_id="goal-a", + status="active", + quota_state="paused", + recommended_action="Wait for the owner.", + ) + status["runtime_root"] = str(tmp_path / "runtime") + status["projection_envelope"] = _status( + _registry(tmp_path, ("goal-a",)), goal_id="goal-a" + )["projection_envelope"] + + paused = build_quota_should_run( + status, + goal_id="goal-a", + runtime_root=status["runtime_root"], + ) + assert paused["state"] == "paused" + assert paused["projection_envelope"]["projection"] == "quota_should_run" + + monkeypatch.setattr( + quota_should_run_module, + "_build_quota_plan_for_goal", + lambda *_args, **_kwargs: ( + { + "ok": False, + "groups": {}, + "health_items": [ + { + "goal_id": "goal-a", + "recommended_action": "Repair the goal state.", + "status": "invalid", + "source": "contract", + } + ], + "summary": {}, + }, + False, + ), + ) + blocked = build_quota_should_run( + status, + goal_id="goal-a", + runtime_root=status["runtime_root"], + ) + assert blocked["state"] == "blocked_health" + assert blocked["projection_envelope"]["projection"] == "quota_should_run" + + +def test_quota_should_run_records_live_workspace_inbox_and_scheduler_reads( + tmp_path: Path, +) -> None: + agent_id = "codex-a" + status = quota_status_payload( + goal_id="goal-a", + status="active", + recommended_action="Advance the goal.", + agent_todo_items=[ + quota_todo_item( + todo_id="todo_delivery", + title="Advance the goal.", + claimed_by=agent_id, + required_capabilities=[], + ) + ], + coordination={"agent_model": "peer_v1", "registered_agents": [agent_id]}, + goal_extra={ + "repo": str(tmp_path), + "control_plane": { + "lark_event_inbox": { + "enabled": True, + "config_path": ".loopx/config/lark/inbox.json", + } + }, + }, + ) + runtime_root = tmp_path / "runtime" + runtime_root.mkdir() + registry_path = tmp_path / "registry.fixture.json" + registry_path.write_text( + json.dumps( + { + "goals": [ + { + "id": "goal-a", + "status": "active", + "repo": str(tmp_path), + "control_plane": { + "lark_event_inbox": { + "enabled": True, + "config_path": ".loopx/config/lark/inbox.json", + } + }, + } + ] + } + ), + encoding="utf-8", + ) + status["registry"] = str(registry_path) + status["runtime_root"] = str(runtime_root) + status["projection_envelope"] = _status( + _registry(tmp_path, ("goal-a",)), goal_id="goal-a" + )["projection_envelope"] + + payload = build_quota_should_run( + status, + goal_id="goal-a", + agent_id=agent_id, + runtime_root=runtime_root, + scheduler_execution_context={ + "host_surface": "codex_app", + "scheduler_owner": "host_automation", + "execution_mode": "hosted_automation", + }, + operator_inbox_urgency_projector=lambda **_kwargs: { + "schema_version": "operator_inbox_urgency_v0", + "enabled": True, + "pending_count": 0, + "attention_required_count": 0, + "reply_due": False, + "material_review_count": 0, + "material_attachment_count": 0, + "material_review_due": False, + "material_review_drain_limit": 20, + "local_private_content_returned": False, + }, + ) + + sources = _sources(payload["projection_envelope"]) + assert sources["operator_inbox"]["status"] == "fresh" + assert sources["delivery_workspace"]["status"] == "fresh" + assert sources["app_scheduler_state"]["status"] == "fresh" + assert str(tmp_path) not in json.dumps(payload["projection_envelope"]) + + +def test_quota_should_run_maps_cadence_read_failures_to_source_status( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + status = _status(_registry(tmp_path, ("goal-a",)), goal_id="goal-a") + original_effect_runtime_result = should_run_packet.effect_runtime_result + + def fail_cadence_read(method: str, *args, **kwargs): + if method == "quota.automation_cadence.manage": + raise RuntimeError("cadence owner unavailable") + return original_effect_runtime_result(method, *args, **kwargs) + + monkeypatch.setattr( + should_run_packet, + "effect_runtime_result", + fail_cadence_read, + ) + failed = build_quota_should_run( + status, + goal_id="goal-a", + runtime_root=status["runtime_root"], + ) + assert _sources(failed["projection_envelope"])["automation_cadence"][ + "status" + ] == "unreadable" + + unavailable_status = quota_status_payload( + goal_id="goal-a", + status="active", + recommended_action="Advance the goal.", + ) + unavailable_status["projection_envelope"] = status["projection_envelope"] + unavailable = build_quota_should_run( + unavailable_status, + goal_id="goal-a", + ) + assert _sources(unavailable["projection_envelope"])["automation_cadence"][ + "status" + ] == "not_read" + + +def test_quota_should_run_inherits_a_replayed_status_envelope( + tmp_path: Path, +) -> None: + registry_path = _registry(tmp_path, ("goal-a",)) + status = _status(registry_path) + args = _cache_args(registry_path) + write_status_projection_cache(payload=status, max_age_seconds=3600, **args) + cached, metadata = load_status_projection_cache( + max_age_seconds=3600, + **args, + ) + + assert metadata["hit"] is True + assert cached["projection_envelope"]["served_from_cache"] is True + payload = build_quota_should_run( + cached, + goal_id="goal-a", + runtime_root=cached["runtime_root"], + ) + + upstream = payload["projection_envelope"]["upstream"] + assert upstream[0]["projection"] == "status" + assert ( + upstream[0]["observed_at"] + == cached["projection_envelope"]["observed_at"] + ) + assert any( + row.get("via") == "status" + for row in payload["projection_envelope"]["sources"] + ) + + +def test_real_cli_quota_should_run_emits_the_projection_envelope( + tmp_path: Path, +) -> None: + import subprocess + import sys + + registry_path = _registry(tmp_path, ("goal-a",)) + completed = subprocess.run( + [ + sys.executable, + "-m", + "loopx.cli", + "--format", + "json", + "--registry", + str(registry_path), + "quota", + "should-run", + "--goal-id", + "goal-a", + ], + cwd=Path(__file__).resolve().parents[1], + capture_output=True, + text=True, + timeout=60, + ) + + assert completed.returncode == 0, completed.stdout + completed.stderr + payload = json.loads(completed.stdout) + assert payload["projection_envelope"]["projection"] == "quota_should_run" + assert payload["projection_envelope"]["coverage"]["included_count"] == 1 + assert payload["projection_envelope"]["upstream"][0]["projection"] == "status" + + def _cache_args(registry_path: Path) -> dict: return { "registry_path": registry_path, From a74b460d9c2355e73ea613b5da1b6757172e8714 Mon Sep 17 00:00:00 2001 From: duang777 Date: Sat, 17 Oct 2026 09:22:51 +0800 Subject: [PATCH 2/4] fix(quota): disclose scheduler state read failures Signed-off-by: duang777 --- loopx/control_plane/quota/should_run.py | 45 ++++++++---- .../control_plane/quota/should_run_packet.py | 8 ++- loopx/control_plane/scheduler/state.py | 11 ++- loopx/control_plane/scheduler/state_store.ts | 69 ++++++++++++++----- 4 files changed, 100 insertions(+), 33 deletions(-) diff --git a/loopx/control_plane/quota/should_run.py b/loopx/control_plane/quota/should_run.py index cfc4db36b1..cd23e4a0be 100644 --- a/loopx/control_plane/quota/should_run.py +++ b/loopx/control_plane/quota/should_run.py @@ -186,9 +186,30 @@ def _attach_quota_should_run_envelope( return payload -def _record_app_scheduler_state_source( - payload: Mapping[str, Any], - status_payload: Mapping[str, Any], +def _app_scheduler_state_read_observer( + direct_sources: dict[str, dict[str, Any]], +) -> Callable[[Mapping[str, Any]], None]: + def record(readback: Mapping[str, Any]) -> None: + previous = direct_sources.get("app_scheduler_state", {}) + unreadable_count = int(previous.get("unreadable_count") or 0) + if readback.get("read_status") == "unreadable": + unreadable_count += 1 + item_count = max( + int(previous.get("item_count") or 0), + 1 if isinstance(readback.get("state"), Mapping) else 0, + ) + direct_sources["app_scheduler_state"] = source_fact( + "app_scheduler_state", + read_status="unreadable" if unreadable_count else "read", + last_read_at=now_utc_iso(), + item_count=item_count, + unreadable_count=unreadable_count, + ) + + return record + + +def _record_unread_app_scheduler_state_source( resolved_scheduler_context: SchedulerExecutionContextResolution, direct_sources: dict[str, dict[str, Any]], ) -> None: @@ -197,19 +218,12 @@ def _record_app_scheduler_state_source( resolved_scheduler_context.ok and context is not None and context.app_automation_applicable + and "app_scheduler_state" not in direct_sources ): return - identity = payload.get("agent_identity") - agent_id = ( - str(identity.get("agent_id") or "").strip() - if isinstance(identity, Mapping) - else "" - ) - read = bool(status_payload.get("runtime_root") and agent_id) direct_sources["app_scheduler_state"] = source_fact( "app_scheduler_state", - read_status="read" if read else "not_read", - last_read_at=now_utc_iso() if read else None, + read_status="not_read", ) @@ -591,10 +605,11 @@ def build_quota_should_run( include_agent_todo_detail=include_agent_todo_detail, runtime_root=runtime_root, goal_ref=goal_ref, + scheduler_state_read_observer=_app_scheduler_state_read_observer( + direct_sources + ), ) - _record_app_scheduler_state_source( - payload, - status_payload, + _record_unread_app_scheduler_state_source( resolved_scheduler_context, direct_sources, ) diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index d354005a94..bd7dc7f1a6 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -2,7 +2,7 @@ from .effective_action import EffectiveAction from ..effect_runtime import effect_runtime_result from .effect_program import ReceiptBoundReplayPhase -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass from pathlib import Path from typing import Any @@ -219,6 +219,7 @@ def _load_app_automation_scheduler_state( goal_id: str, agent_id: str | None, surface: str, + read_observer: Callable[[Mapping[str, Any]], None] | None = None, ) -> dict[str, Any] | None: raw_runtime_root = status_payload.get("runtime_root") safe_agent_id = normalize_todo_claimed_by(agent_id) @@ -229,6 +230,7 @@ def _load_app_automation_scheduler_state( goal_id=goal_id, agent_id=safe_agent_id, surface=surface, + read_observer=read_observer, ) @@ -1628,6 +1630,9 @@ def _build_quota_should_run_payload( include_agent_todo_detail: bool = False, runtime_root: str | Path | None = None, goal_ref: Mapping[str, object] | None = None, + scheduler_state_read_observer: ( + Callable[[Mapping[str, Any]], None] | None + ) = None, ) -> dict[str, Any]: if prepared.receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED: payload = _build_settled_quota_payload(prepared, route) @@ -1669,6 +1674,7 @@ def _build_quota_should_run_payload( agent_id=quota_decision_agent_id(payload) or prepared.requested_agent_id, surface=prepared.resolved_scheduler_context.context.host_surface.value, + read_observer=scheduler_state_read_observer, ) if prepared.resolved_scheduler_context.ok and prepared.resolved_scheduler_context.context is not None diff --git a/loopx/control_plane/scheduler/state.py b/loopx/control_plane/scheduler/state.py index 5c669b295a..c00c5a6d84 100644 --- a/loopx/control_plane/scheduler/state.py +++ b/loopx/control_plane/scheduler/state.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Callable, Mapping from datetime import datetime from pathlib import Path from typing import Any @@ -30,6 +30,7 @@ LEGACY_CODEX_APP_STATEFUL_BACKOFF_STATE_KEY ) CODEX_APP_SURFACE = "codex_app" +SchedulerStateReadObserver = Callable[[Mapping[str, Any]], None] def _operation_result(operation: str, **params: Any) -> Any: @@ -285,6 +286,7 @@ def load_scheduler_state( agent_id: str | None, surface: str = CODEX_APP_SURFACE, state_key: str = CODEX_APP_STATEFUL_BACKOFF_STATE_KEY, + read_observer: SchedulerStateReadObserver | None = None, ) -> dict[str, Any] | None: if not agent_id: return None @@ -300,6 +302,10 @@ def load_scheduler_state( state = result.get("state") if state is not None and not isinstance(state, dict): raise RuntimeError("TypeScript loaded scheduler state must be an object or null") + if result.get("read_status") not in {"read", "unreadable"}: + raise RuntimeError("TypeScript scheduler state read status is invalid") + if read_observer is not None: + read_observer(result) return state @@ -309,6 +315,7 @@ def load_app_automation_scheduler_state( goal_id: str, agent_id: str | None, surface: str, + read_observer: SchedulerStateReadObserver | None = None, ) -> dict[str, Any] | None: current = load_scheduler_state( runtime_root, @@ -316,6 +323,7 @@ def load_app_automation_scheduler_state( agent_id=agent_id, surface=surface, state_key=APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY, + read_observer=read_observer, ) if current is not None or surface != CODEX_APP_SURFACE: return current @@ -328,6 +336,7 @@ def load_app_automation_scheduler_state( agent_id=agent_id, surface=surface, state_key=LEGACY_CODEX_APP_STATEFUL_BACKOFF_STATE_KEY, + read_observer=read_observer, ) diff --git a/loopx/control_plane/scheduler/state_store.ts b/loopx/control_plane/scheduler/state_store.ts index d57a0bd10b..53d498034b 100644 --- a/loopx/control_plane/scheduler/state_store.ts +++ b/loopx/control_plane/scheduler/state_store.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import type { SourceReadStatus } from "../projection_envelope.ts"; import { atomicWriteJson, withFileMutationLock, @@ -75,6 +76,14 @@ export interface SchedulerStateLoadResult { operation: "load"; path: string; state: JsonObject | null; + read_status: SchedulerStateReadStatus; +} + +type SchedulerStateReadStatus = Extract; + +interface SchedulerStateReadResult { + state: JsonObject | null; + readStatus: SchedulerStateReadStatus; } interface SchedulerStateWriteResultBase { @@ -502,7 +511,7 @@ async function migrateLegacySchedulerStateAtPathsUnderLock( scope: SchedulerScope, path: string, legacyPath: string, -): Promise { +): Promise { // Probe the legacy file before creating any directory or taking the legacy // lock. The legacy layout uses unbounded safeSegment labels, so an overlong // scope component can make the legacy path unaddressable (ENAMETOOLONG) or @@ -513,7 +522,9 @@ async function migrateLegacySchedulerStateAtPathsUnderLock( await readFile(legacyPath, "utf8"); } catch (error) { const code = (error as NodeJS.ErrnoException).code; - if (code === "ENOENT" || code === "ENAMETOOLONG") return null; + if (code === "ENOENT" || code === "ENAMETOOLONG") { + return { state: null, readStatus: "read" }; + } throw error; } return await withFileMutationLock(legacyPath, async () => { @@ -522,7 +533,7 @@ async function migrateLegacySchedulerStateAtPathsUnderLock( JSON.parse(await readFile(path, "utf8")), scope, ); - if (current) return current; + if (current) return { state: current, readStatus: "read" }; } catch { // The canonical path is still absent or unusable; inspect legacy state. } @@ -532,17 +543,23 @@ async function migrateLegacySchedulerStateAtPathsUnderLock( JSON.parse(await readFile(legacyPath, "utf8")), scope, ); - } catch { - return null; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + return { + state: null, + readStatus: code === "ENOENT" || code === "ENAMETOOLONG" + ? "read" + : "unreadable", + }; } - if (!legacyState) return null; + if (!legacyState) return { state: null, readStatus: "unreadable" }; await atomicWriteJson(path, stableValue(legacyState) as JsonObject); try { await unlink(legacyPath); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } - return legacyState; + return { state: legacyState, readStatus: "read" }; }); } @@ -550,7 +567,7 @@ async function migrateLegacySchedulerState( scope: SchedulerScope, path: string, legacyPath: string, -): Promise { +): Promise { return await withFileMutationLock( path, async () => @@ -566,26 +583,34 @@ export async function migrateLegacySchedulerStateUnderLock( runtimeRoot: string, scope: SchedulerScope, ): Promise { - return await migrateLegacySchedulerStateAtPathsUnderLock( + const result = await migrateLegacySchedulerStateAtPathsUnderLock( scope, schedulerStatePath(runtimeRoot, scope), legacySchedulerStatePath(runtimeRoot, scope), ); + return result.state; } async function readLegacySchedulerState( scope: SchedulerScope, legacyPath: string, -): Promise { +): Promise { try { - return normalizeSchedulerState( + const state = normalizeSchedulerState( JSON.parse(await readFile(legacyPath, "utf8")), scope, ); + return { + state, + readStatus: state === null ? "unreadable" : "read", + }; } catch (error) { const code = (error as NodeJS.ErrnoException).code; - if (code === "ENOENT" || code === "ENAMETOOLONG" || error instanceof SyntaxError) { - return null; + if (code === "ENOENT" || code === "ENAMETOOLONG") { + return { state: null, readStatus: "read" }; + } + if (error instanceof SyntaxError) { + return { state: null, readStatus: "unreadable" }; } throw error; } @@ -596,13 +621,24 @@ export async function loadSchedulerState( ): Promise { const { request, scope, path, legacyPath } = storeRequest(value, "load"); let state: JsonObject | null = null; + let readStatus: SchedulerStateReadStatus = "read"; try { state = normalizeSchedulerState(JSON.parse(await readFile(path, "utf8")), scope); + if (state === null) readStatus = "unreadable"; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") { - state = request.migrate_legacy === false - ? await readLegacySchedulerState(scope, legacyPath) - : await migrateLegacySchedulerState(scope, path, legacyPath); + try { + const legacy = request.migrate_legacy === false + ? await readLegacySchedulerState(scope, legacyPath) + : await migrateLegacySchedulerState(scope, path, legacyPath); + state = legacy.state; + readStatus = legacy.readStatus; + } catch { + readStatus = "unreadable"; + state = null; + } + } else { + readStatus = "unreadable"; } } return { @@ -610,6 +646,7 @@ export async function loadSchedulerState( operation: "load", path, state, + read_status: readStatus, }; } From 7f6c8c7248b20d3086a6bccb04661dfcf991f76f Mon Sep 17 00:00:00 2001 From: duang777 Date: Sat, 17 Oct 2026 09:22:59 +0800 Subject: [PATCH 3/4] test(quota): cover unreadable scheduler state Signed-off-by: duang777 --- .../scheduler_state_store.test.ts | 49 +++++++++- tests/test_projection_envelope.py | 90 +++++++++++++++++++ 2 files changed, 137 insertions(+), 2 deletions(-) diff --git a/tests/control_plane_ts/scheduler_state_store.test.ts b/tests/control_plane_ts/scheduler_state_store.test.ts index 030962569b..01cd69dca3 100644 --- a/tests/control_plane_ts/scheduler_state_store.test.ts +++ b/tests/control_plane_ts/scheduler_state_store.test.ts @@ -310,13 +310,58 @@ test("state store writes atomically and recognizes replay", async (t) => { ); }); -test("malformed persisted JSON is a missing state, not partial truth", async (t) => { +test("a missing state is a successful empty read", async (t) => { + const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-")); + t.after(() => rm(runtimeRoot, { recursive: true, force: true })); + + const result = await loadSchedulerState(storeRequest(runtimeRoot)); + + assert.equal(result.state, null); + assert.equal(result.read_status, "read"); +}); + +test("malformed persisted JSON is unreadable, not partial truth", async (t) => { const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-")); t.after(() => rm(runtimeRoot, { recursive: true, force: true })); const path = schedulerStatePath(runtimeRoot, scope); await writeSchedulerState(storeRequest(runtimeRoot, { state: state() })); await writeFile(path, "{truncated", "utf8"); - assert.equal((await loadSchedulerState(storeRequest(runtimeRoot))).state, null); + const result = await loadSchedulerState(storeRequest(runtimeRoot)); + assert.equal(result.state, null); + assert.equal(result.read_status, "unreadable"); +}); + +test("malformed legacy persisted JSON is unreadable, not missing", async (t) => { + const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-")); + t.after(() => rm(runtimeRoot, { recursive: true, force: true })); + const legacyPath = join( + runtimeRoot, + "goals", + scope.goalId, + "scheduler-state", + scope.agentId, + scope.surface, + "8a41f410c67e7c0e.json", + ); + await mkdir(dirname(legacyPath), { recursive: true }); + await writeFile(legacyPath, "{truncated", "utf8"); + + const result = await loadSchedulerState(storeRequest(runtimeRoot)); + + assert.equal(result.state, null); + assert.equal(result.read_status, "unreadable"); +}); + +test("a directory at the persisted state path is unreadable", async (t) => { + const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-")); + t.after(() => rm(runtimeRoot, { recursive: true, force: true })); + const path = schedulerStatePath(runtimeRoot, scope); + await mkdir(path, { recursive: true }); + + const result = await loadSchedulerState(storeRequest(runtimeRoot)); + + assert.equal(result.state, null); + assert.equal(result.read_status, "unreadable"); }); test("concurrent same-key writes serialize without partial state", async (t) => { diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py index 189fb578c6..4548a1e080 100644 --- a/tests/test_projection_envelope.py +++ b/tests/test_projection_envelope.py @@ -23,6 +23,12 @@ write_status_projection_cache, ) from loopx.control_plane.runtime.time import now_utc, utc_isoformat +from loopx.control_plane.scheduler.state import ( + APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY, + CODEX_APP_SURFACE, + LEGACY_CODEX_APP_STATEFUL_BACKOFF_STATE_KEY, + scheduler_state_path, +) from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown from loopx.presentation.renderers.status_markdown import render_status_markdown from loopx.quota import build_quota_should_run @@ -466,6 +472,90 @@ def test_real_cli_quota_should_run_emits_the_projection_envelope( assert payload["projection_envelope"]["upstream"][0]["projection"] == "status" +def test_real_cli_quota_should_run_discloses_unreadable_scheduler_state( + tmp_path: Path, +) -> None: + import subprocess + import sys + + registry_path = _registry(tmp_path, ("goal-a",)) + runtime_root = tmp_path / "runtime" + agent_id = "reader-a" + registry = json.loads(registry_path.read_text(encoding="utf-8")) + registry["goals"][0]["coordination"] = { + "agent_model": "peer_v1", + "registered_agents": [agent_id], + } + registry_path.write_text(json.dumps(registry), encoding="utf-8") + command = [ + sys.executable, + "-m", + "loopx.cli", + "--format", + "json", + "--registry", + str(registry_path), + "quota", + "should-run", + "--runtime-profile", + "codex_app_heartbeat", + "--goal-id", + "goal-a", + "--agent-id", + agent_id, + ] + + def run_cli() -> dict: + completed = subprocess.run( + command, + cwd=Path(__file__).resolve().parents[1], + capture_output=True, + text=True, + timeout=60, + ) + assert completed.returncode == 0, completed.stdout + completed.stderr + return json.loads(completed.stdout) + + missing = _sources(run_cli()["projection_envelope"])["app_scheduler_state"] + assert missing["read_status"] == "read" + assert missing["status"] == "fresh" + + paths = [ + scheduler_state_path( + runtime_root, + goal_id="goal-a", + agent_id=agent_id, + surface=CODEX_APP_SURFACE, + state_key=state_key, + ) + for state_key in ( + APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY, + LEGACY_CODEX_APP_STATEFUL_BACKOFF_STATE_KEY, + ) + ] + for path in paths: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("{truncated", encoding="utf-8") + + corrupt_payload = run_cli() + corrupt = _sources(corrupt_payload["projection_envelope"])[ + "app_scheduler_state" + ] + assert corrupt["read_status"] == "unreadable" + assert corrupt["status"] == "unreadable" + assert corrupt_payload["projection_envelope"]["alert"] is True + + for path in paths: + path.unlink() + recovered_payload = run_cli() + recovered = _sources(recovered_payload["projection_envelope"])[ + "app_scheduler_state" + ] + assert recovered["read_status"] == "read" + assert recovered["status"] == "fresh" + assert recovered_payload["projection_envelope"]["alert"] is False + + def _cache_args(registry_path: Path) -> dict: return { "registry_path": registry_path, From 9ef61099af9ee83772020afb052e4d180feff6c3 Mon Sep 17 00:00:00 2001 From: duang777 Date: Sat, 17 Oct 2026 09:26:56 +0800 Subject: [PATCH 4/4] test(quota): exercise scheduler state IO failure Signed-off-by: duang777 --- tests/test_projection_envelope.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py index 4548a1e080..c8c6b0d787 100644 --- a/tests/test_projection_envelope.py +++ b/tests/test_projection_envelope.py @@ -547,6 +547,17 @@ def run_cli() -> dict: for path in paths: path.unlink() + path.mkdir() + directory_payload = run_cli() + directory = _sources(directory_payload["projection_envelope"])[ + "app_scheduler_state" + ] + assert directory["read_status"] == "unreadable" + assert directory["status"] == "unreadable" + assert directory_payload["projection_envelope"]["alert"] is True + + for path in paths: + path.rmdir() recovered_payload = run_cli() recovered = _sources(recovered_payload["projection_envelope"])[ "app_scheduler_state"