From d9d7978f1e59d87fbe7edcfd1ec47cb4d4ad1ce2 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 26 Sep 2026 04:08:36 +0800 Subject: [PATCH 1/4] feat(projection): kernel-owned projection envelope for status and global views Add loopx_projection_envelope_v0, sealed by the TypeScript kernel through projection.envelope.seal. Python adapters pass compact read facts only; TS decides freshness, alerts, and completeness. - status / --goal-id: per-source last_read_at and coverage relative to the requested scope; cached copies keep observed_at and restamp served_at - global-summary / global-gates: goal_quota source, upstream status envelope, and outside_current_registry omissions - Markdown renders a red projection line when stale, unreadable, missing, or incomplete - RFC: TypeScript control-plane migration section 2.6, baseline row, and correctness rule; reference contract and status data contract Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Co-authored-by: Cursor Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Co-authored-by: Cursor --- .../typescript-control-plane-migration-v0.md | 52 ++- ...script-control-plane-migration-v0.zh-CN.md | 43 ++- docs/reference/contracts/README.md | 1 + .../contracts/projection-envelope-contract.md | 125 +++++++ docs/status-data-contract.md | 15 + .../control_plane/effect_runtime_handlers.ts | 2 + .../goals/global_registry_health.py | 2 + loopx/control_plane/projection_envelope.ts | 325 ++++++++++++++++++ .../projection_envelope_facts.py | 115 +++++++ .../runtime/status_projection_cache.py | 13 + loopx/control_plane/status/collection.py | 100 ++++++ .../presentation/renderers/status_markdown.py | 2 + .../project_registry_io_manifest_v1.json | 4 +- loopx/summary_all.py | 100 +++++- .../projection_envelope.test.ts | 182 ++++++++++ tests/test_projection_envelope.py | 259 ++++++++++++++ 16 files changed, 1334 insertions(+), 6 deletions(-) create mode 100644 docs/reference/contracts/projection-envelope-contract.md create mode 100644 loopx/control_plane/projection_envelope.ts create mode 100644 loopx/control_plane/projection_envelope_facts.py create mode 100644 tests/control_plane_ts/projection_envelope.test.ts create mode 100644 tests/test_projection_envelope.py diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 26c7d5e205..b245d91fc2 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -3,7 +3,7 @@ - Status: Accepted, transaction-payoff phase in progress - Proposed by: LoopX maintainers - Date: 2026-08-15 -- Last revised: 2026-09-13 +- Last revised: 2026-09-26 - Scope: an incremental, replacement-first migration of the LoopX control-plane core from Python to TypeScript without maintaining two semantic implementations @@ -1407,6 +1407,52 @@ while public, persisted, RPC, or extension input still reaches its semantic core through an unvalidated assertion. TypeScript complements runtime validation; it does not replace it. +### 2.6 Projection envelope is a kernel read contract + +The kernel already binds writes to receipts, fences and CAS. Reads need a +matching contract. A read model such as status, a global summary or a context +packet combines several sources read at different times, and it is often +consumed later from a cache, a saved file or a pasted packet. Without a +machine-checkable statement of what it observed, consumers, agents above all, +treat an old or partial projection as the current whole state. `ok: true`, a +passing check or a healthy host says nothing about that. + +Every operator- or agent-facing projection therefore carries one +`projection_envelope` (`loopx_projection_envelope_v0`): + +- `observed_at` and `served_at`: when the sources were read and when this copy + was emitted. A cache hit or replay keeps the first and restamps the second. +- One row per source with `last_read_at`, `read_status`, window, staleness and + alert reasons. A derived projection inherits its upstream rows, so it cannot + look fresher than the oldest read it depends on. +- `coverage` of the requested scope: expected and included counts, and named + omissions. Display truncation is disclosed separately and is not an + incompleteness alert. + +Ownership follows this RFC instead of creating new migration debt. +`projection_envelope.ts` alone decodes the facts and decides staleness, alerts +and completeness, through runtime method `projection.envelope.seal`. +Python-owned projections only pass compact read facts. That is one request per +projection, on paths that already pin a runtime revision and make dozens of TS +calls. It is not a leaf migration: it keeps a new cross-cutting rule from +being born in Python and migrated later. The Python facts adapter exits with +its projection: when status projection moves into the kernel (already a +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` and `global-gates` 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: `global-todos` and `global-risks` (the +same composition, one call each), `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. + +Consumers treat a missing envelope as unknown freshness, and disclose an +alerting one before stating any conclusion that depends on it. Field +semantics and the consumer rule are in the +[projection envelope contract](../../reference/contracts/projection-envelope-contract.md). + ## 3. Current baseline and phase transition Effect Program moved first because it joins ordered steps, identity, @@ -1426,6 +1472,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/huangruiteng/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/huangruiteng/loopx/pull/3464), [#3481](https://github.com/huangruiteng/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/event 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` and `global-gates` 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 @@ -1794,6 +1841,9 @@ not authorize a generic schema framework. concurrent same-key mutations are serialized or use a tested CAS contract, and retry identity distinguishes successive checkpoints within one Turn. - Process crash and retry cannot duplicate a committed internal effect. +- An operator- or agent-facing read model that is added or migrated emits + `projection_envelope` through `projection.envelope.seal`. Its tests cover a + stale source, an unreadable source, an incomplete scope and a replayed copy. - Wheel and sdist are installed into fresh environments and execute deep semantic probes from packaged files. 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 f84b005a69..4ba6b910a2 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 @@ -3,7 +3,7 @@ - Status：Accepted，transaction-payoff 阶段进行中 - Proposed by：LoopX maintainers - Date：2026-08-15 -- Last revised：2026-09-13 +- Last revised：2026-09-26 - Scope：LoopX 控制面核心从 Python 到 TypeScript 的增量、replacement-first 迁移；不长期维护两份语义实现 - Tracking issue：[#3225](https://github.com/huangruiteng/loopx/issues/3225) @@ -1049,6 +1049,43 @@ validator、负向边界覆盖和移除 owner。只要 public、持久化、RPC 输入仍通过未经验证的断言进入已迁 domain 的 semantic core，该 domain 就不能 通过 promotion gate。TypeScript 补充运行时验证，而不是替代它。 +### 2.6 Projection envelope 是 kernel 级读合同 + +kernel 已经用 receipt、fence 与 CAS 约束写入，读取也需要对应的合同。status、 +全局摘要、context packet 这类读模型由多个在不同时间读取的来源组合而成，而且常常 +在事后通过 cache、保存的文件或粘贴的 packet 被消费。如果没有机器可检查的"它看到 +了什么"，消费者（尤其是 agent）会把旧的或不完整的投影当作当前的全貌。`ok: true`、 +检查通过或 host 健康都不说明这一点。 + +因此每个面向 operator 或 agent 的投影都携带一个 `projection_envelope` +（`loopx_projection_envelope_v0`）： + +- `observed_at` 与 `served_at`：来源何时被读取、这份副本何时被输出。cache 命中或 + 重放保留前者、重盖后者。 +- 每个来源一行，含 `last_read_at`、`read_status`、窗口、staleness 与告警原因。 + 派生投影继承上游的来源行，因此不可能显得比它依赖的最旧读取更新。 +- 对所请求范围的 `coverage`：期望数与已包含数，以及具名的缺漏。显示截断单独披露， + 不算不完整告警。 + +归属遵循本 RFC，而不是制造新的迁移债务。只有 `projection_envelope.ts` 解码这些 +facts，并决定 staleness、告警与完整性，runtime 方法为 `projection.envelope.seal`。 +Python 拥有的投影只传入紧凑的读取 facts。每个投影一次请求，而这些路径本来就固定 +了 runtime revision、每次要发起几十次 TS 调用。这不是 leaf 迁移：它避免一条新的 +横切规则先在 Python 里诞生、之后再迁移。Python facts adapter 随其投影退出：当 +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 门禁。 + +消费者把缺失 envelope 视为新鲜度未知；envelope 告警时，必须先披露，再陈述依赖它的 +结论。字段语义与消费者规则见 +[projection envelope 合同](../../reference/contracts/projection-envelope-contract.md)（仅英文）。 + ## 3. 当前基线与阶段转换 Effect Program 先迁，是因为它连接 ordered step、identity、short-circuit failure、 @@ -1067,6 +1104,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/huangruiteng/loopx/pull/3443)） | 稳定 primitive decoding 进入一个很小的共享模块；domain decoder 仍留在本地 | 没有理由建设更大的 schema framework | | Transaction 兑现（[#3464](https://github.com/huangruiteng/loopx/pull/3464)、[#3481](https://github.com/huangruiteng/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/event result；其他 domain 仍需各自的 bounded cutover | +| Projection envelope | TypeScript 拥有 `loopx_projection_envelope_v0` 的解码，以及全部 freshness、告警、完整性与重放判定 | 在 `status`、`global-summary`、`global-gates` 迁移前，Python 仍为它们收集读取 facts | Scheduler facade exit 已交付第一段有边界的 Stage 3 路径。带版本的 `heartbeat_followup_cli.ts` 从生成的 ACK/failure hint 接收有大小上限的 compact host @@ -1382,6 +1420,9 @@ happy path 及其 retry/recovery path 上实测，不能由 handler 数量推断 并发 mutation 必须串行化或使用经过测试的 CAS 合同，retry identity 必须区分同一 Turn 内连续发生的 checkpoint。 - 进程 crash 与 retry 不得重复已经提交的内部 effect。 +- 新增或迁移的、面向 operator 或 agent 的读模型通过 `projection.envelope.seal` + 输出 `projection_envelope`；其测试覆盖 stale 来源、不可读来源、不完整范围与 + 重放副本。 - wheel 与 sdist 安装到全新环境后，从打包文件执行 deep semantic probe。 #### Caller 可观测语义是 promotion 门禁 diff --git a/docs/reference/contracts/README.md b/docs/reference/contracts/README.md index 96f7a4a4b7..15ee948b84 100644 --- a/docs/reference/contracts/README.md +++ b/docs/reference/contracts/README.md @@ -7,6 +7,7 @@ implementation modules. - [Dashboard budget governance](dashboard-budget-governance-contract.md) - [Dashboard reward write boundary](dashboard-reward-write-boundary.md) - [Reward gate direct-write contract](reward-gate-direct-write-contract.md) +- [Projection envelope contract](projection-envelope-contract.md) - [Status data contract](../../status-data-contract.md) - [Quota allocation](../../quota-allocation.md) - [Project agent todo contract](../../project-agent-todo-contract.md) diff --git a/docs/reference/contracts/projection-envelope-contract.md b/docs/reference/contracts/projection-envelope-contract.md new file mode 100644 index 0000000000..1167459d4e --- /dev/null +++ b/docs/reference/contracts/projection-envelope-contract.md @@ -0,0 +1,125 @@ +# Projection Envelope Contract + +Every operator- or agent-facing read model carries one `projection_envelope` +(`loopx_projection_envelope_v0`). It says when the projection observed its +sources, how old each source read is at the moment this copy is served, and +how much of the requested scope the projection covers. The rule that +[RFC §2.6](../../architecture/rfcs/typescript-control-plane-migration-v0.md#26-projection-envelope-is-a-kernel-read-contract) +makes kernel-owned is simple: `ok: true`, a passing check, a healthy host, or +a cache hit never implies fresh or complete sources. + +The TypeScript kernel (`loopx/control_plane/projection_envelope.ts`, runtime +method `projection.envelope.seal`) is the only implementation that decides +freshness, alerts, and completeness. Producers pass compact read facts; +Python-owned projections do so through `loopx/control_plane/projection_envelope_facts.py`. + +## Current carriers + +| Projection | `projection` | Coverage scope | +| --- | --- | --- | +| `loopx status` | `status` | `registry`, `goal` with `--goal-id`, or `activation.` | +| `loopx status --use-projection-cache` hit | `status`, `served_from_cache: true` | as stored | +| `loopx global-summary` | `global_summary` | `global` | +| `loopx global-gates` | `global_gates` | `global` | + +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. + +## Shape + +```json +{ + "schema_version": "loopx_projection_envelope_v0", + "projection": "global_summary", + "observed_at": "2026-09-26T10:00:05Z", + "served_at": "2026-09-26T10:00:05Z", + "age_seconds": 0, + "served_from_cache": false, + "fresh": true, + "complete": false, + "alert": true, + "alert_reasons": ["incomplete_coverage"], + "alert_source_ids": [], + "sources": [ + { + "source_id": "goal_quota", + "required": true, + "read_status": "read", + "last_read_at": "2026-09-26T10:00:04Z", + "source_updated_at": null, + "window_seconds": 300, + "staleness_seconds": 1, + "status": "fresh", + "item_count": 1, + "missing_count": 0, + "unreadable_count": 0, + "alert": false, + "alert_reasons": [] + }, + { "source_id": "registry", "via": "status", "...": "inherited from the status envelope" } + ], + "coverage": { + "scope": "global", + "expected_count": 48, + "included_count": 1, + "omitted": [{ "reason": "outside_current_registry", "count": 47, "refs": ["goal-b", "..."] }], + "shown_count": 8, + "available_count": 19, + "truncated": true, + "complete": false + }, + "upstream": [{ "projection": "status", "observed_at": "2026-09-26T10:00:03Z", "complete": true }] +} +``` + +## Field semantics + +- `observed_at`: when the projection finished reading its sources. A cached or + persisted copy keeps it; `served_at` and `age_seconds` say when this copy + was emitted. +- `last_read_at`: when this projection (or its upstream) last read the source. + `staleness_seconds = served_at - last_read_at`; a read source is `stale` once + that exceeds `window_seconds` (kernel default 300). +- `source_updated_at`: the source's own last write when the producer knows it, + such as the newest run in the run indexes. An old value means the goal has + not moved, not that the read is stale; it never raises an alert. +- `read_status` is `read`, `missing`, `unreadable`, or `not_read`. `unreadable` + always alerts; `missing` and `not_read` alert only when `required` is true. + A read aggregate with `unreadable_count > 0` alerts as `partially_unreadable`. +- `coverage` is relative to the requested scope. `complete` requires no + `omitted` rows, `included_count >= expected_count`, and every upstream + envelope complete. `truncated` only discloses a requested display limit and + never alerts on its own. +- `via` marks a row inherited from an upstream envelope. A derived projection + inherits every upstream source row, so its freshness is bounded by the oldest + read it depends on, not by its own assembly time. +- Envelope-level `alert_reasons` is a sorted subset of `stale_sources`, + `unreadable_sources`, `missing_required_sources`, and `incomplete_coverage`. + +Source ids, scope names, and reason codes are lowercase identifiers. The +envelope never contains filesystem paths, so public-safe projections can carry +it unchanged. Omission `refs` hold at most 8 identifiers such as goal ids. + +## Consumer rule + +Before stating that something is the current or whole state, a consumer reads +the envelope and discloses it: + +1. A missing envelope means freshness and coverage are unknown. Say so; do not + infer freshness from `ok`, `generated_at`, or a recent command. +2. `alert: true` must be stated along with the reasons, the affected + `alert_source_ids`, and the omissions before any conclusion that depends on + them. Markdown renderers print this as a `🔴 projection alerts` line. +3. A copy read later than `served_at` (a pasted packet, a saved file, a chat + quote) is as old as `now - observed_at`. Re-read the projection instead of + re-serving it from memory. +4. `truncated: true` means only `shown_count` of `available_count` items are + listed. Absence from the list is not absence from the scope. + +## Cache and replay + +`status --use-projection-cache` re-serves the stored envelope through the +kernel on every hit, which restamps `served_at` and recomputes staleness. A +cache record without an envelope, or with one the kernel decoder rejects, is a +cache miss (`missing_projection_envelope` / `invalid_projection_envelope`), +never an unlabeled hit. diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index 6aa8e6976d..568a4f2ca6 100644 --- a/docs/status-data-contract.md +++ b/docs/status-data-contract.md @@ -328,6 +328,8 @@ goals must stay out of the eligible lane even when they have a high "current_registry_is_global": false, "global_goal_count": 4, "current_goal_count": 3, + "current_registry_excluded_goal_count": 1, + "current_registry_excluded_goal_ids": ["other-project-goal"], "source_registry_count": 2, "summary": { "high": 0, @@ -2244,6 +2246,19 @@ release artifact under the LoopX runtime root. non-blocking warning; operators should still run `loopx doctor` or the canary-promotion readiness smoke for exact local release evidence. +## Projection Envelope + +`projection_envelope` (`loopx_projection_envelope_v0`) records when status read +the registry, global registry, run indexes, goal state contract, and runtime +projection routes, and whether the requested scope (`registry`, `goal`, or +`activation.`) is fully covered. Registry members count toward +coverage; legacy runtime goals are extra. An unknown `--goal-id` is reported as +`goal_not_found` rather than as an empty but complete projection. A +`--use-projection-cache` hit re-serves the stored envelope, keeping +`observed_at` and restamping `served_at` and staleness. Field semantics and +the consumer rule are in the +[projection envelope contract](reference/contracts/projection-envelope-contract.md). + ## Decision Freshness Summary `decision_freshness_summary` is an optional checkpointed-decision projection over diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 730c3b9c8a..a774751aff 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -13,6 +13,7 @@ import {evaluateTodoPriority} from "./todos/priority.ts"; import {evaluateUserCompletion} from "./todos/user_completion.ts"; import {projectTodoSuccession} from "./todos/succession.ts"; import {projectLegacyTodoWorkCounts} from "./todos/summary_lanes.ts"; +import {sealProjectionEnvelope} from "./projection_envelope.ts"; import {recordDelegationAdoption, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, recoverValidatedDelegationSettlement, selectDelegationBinding, transitionDelegationObservation} from "./collaboration/delegation.ts"; import {planChatMode} from "./collaboration/chat_mode.ts"; import {resolveConversationScope} from "./collaboration/conversation_scope.ts"; @@ -453,6 +454,7 @@ export function createEffectRuntimeHandlers( ["capabilities.periodic_report.approval_retry.select", selectPeriodicReportApprovalRetry], ["todo.succession.project", projectTodoSuccession], ["todo.work_counts.project", projectLegacyTodoWorkCounts], + ["projection.envelope.seal", sealProjectionEnvelope], ["todo.decision_scope.evaluate", evaluateDecisionScope], ["todo.user_completion.plan", evaluateUserCompletion], ["agent.capability_gate.evaluate", evaluateCapabilityGate], diff --git a/loopx/control_plane/goals/global_registry_health.py b/loopx/control_plane/goals/global_registry_health.py index 004f395429..ce01830549 100644 --- a/loopx/control_plane/goals/global_registry_health.py +++ b/loopx/control_plane/goals/global_registry_health.py @@ -186,6 +186,8 @@ def collect_global_registry_health( "current_registry_is_global": current_is_global, "global_goal_count": len(global_goals), "current_goal_count": len(current_goals), + "current_registry_excluded_goal_count": 0 if current_is_global else len(missing_from_current), + "current_registry_excluded_goal_ids": [] if current_is_global else missing_from_current[:8], "source_registry_count": len(source_registries), "summary": { "high": severity_counts.get("high", 0), diff --git a/loopx/control_plane/projection_envelope.ts b/loopx/control_plane/projection_envelope.ts new file mode 100644 index 0000000000..aae6a1988f --- /dev/null +++ b/loopx/control_plane/projection_envelope.ts @@ -0,0 +1,325 @@ +/** Kernel projection envelope: when a read model observed each source, how + * fresh each read is when the copy is served, and how much of the requested + * scope it covers. `ok`, a passing check, or a cache hit never implies fresh + * or complete sources; consumers read this envelope before stating current + * state. */ +import type { JsonObject } from "./effect_program.ts"; +import { EffectRuntimeRequestError } from "./effect_runtime_errors.ts"; +import { + requireBoolean, + requireInteger, + requireJsonObject, + requireNonEmptyString, + requireStringLiteral, +} from "./runtime_decode.ts"; +import { parseIsoTimestamp } from "./runtime_timestamp.ts"; + +export const PROJECTION_ENVELOPE_SCHEMA_VERSION = "loopx_projection_envelope_v0"; +export const PROJECTION_ENVELOPE_SEAL_REQUEST = "loopx_projection_envelope_seal_request_v0"; +export const PROJECTION_ENVELOPE_SERVE_REQUEST = "loopx_projection_envelope_serve_request_v0"; +export const DEFAULT_SOURCE_WINDOW_SECONDS = 300; +export const SOURCE_READ_STATUSES = ["read", "missing", "unreadable", "not_read"] as const; +export type SourceReadStatus = typeof SOURCE_READ_STATUSES[number]; +export type SourceStatus = "fresh" | "stale" | Exclude; + +const MAX_SOURCES = 32; +const MAX_OMISSIONS = 16; +const MAX_REFS = 8; +const MAX_UPSTREAM = 4; +const MAX_REF_CHARS = 128; +const IDENTIFIER = /^[a-z][a-z0-9_.]{0,63}$/u; +const REASON = /^[a-z][a-z0-9_.:-]{0,63}$/u; +const EXPLICIT_OFFSET = /(?:Z|z|[+-]\d{2}(?::?\d{2})?)$/u; + +interface SourceFact { + source_id: string; + via: string | null; + required: boolean; + read_status: SourceReadStatus; + last_read_at: string | null; + source_updated_at: string | null; + window_seconds: number; + item_count: number | null; + missing_count: number; + unreadable_count: number; +} + +interface Omission { + reason: string; + count: number; + refs: string[]; +} + +interface CoverageFact { + scope: string; + expected_count: number | null; + included_count: number; + omitted: Omission[]; + shown_count: number | null; + available_count: number | null; +} + +interface UpstreamSummary { + projection: string; + observed_at: string; + complete: boolean; +} + +function timestamp(value: unknown, label: string): { text: string; millis: number } { + const text = requireNonEmptyString(value, label).trim(); + const parsed = EXPLICIT_OFFSET.test(text) ? parseIsoTimestamp(text) : null; + if (parsed === null) { + throw new EffectRuntimeRequestError(`${label} must be an ISO timestamp with an explicit offset`); + } + return { text, millis: parsed.getTime() }; +} + +function optionalTimestamp(value: unknown, label: string): string | null { + return value === null || value === undefined ? null : timestamp(value, label).text; +} + +function identifier(value: unknown, label: string): string { + const text = requireNonEmptyString(value, label); + if (!IDENTIFIER.test(text)) throw new EffectRuntimeRequestError(`${label} is not a projection identifier`); + return text; +} + +function count(value: unknown, label: string): number { + const result = requireInteger(value, label); + if (result < 0) throw new EffectRuntimeRequestError(`${label} must be non-negative`); + return result; +} + +function optionalCount(value: unknown, label: string): number | null { + return value === null || value === undefined ? null : count(value, label); +} + +function boundedArray(value: unknown, limit: number, label: string): unknown[] { + if (!Array.isArray(value)) throw new EffectRuntimeRequestError(`${label} must be an array`); + if (value.length > limit) throw new EffectRuntimeRequestError(`${label} exceeds ${limit} entries`); + return value; +} + +function decodeSource(value: unknown, index: number): SourceFact { + const label = `sources[${index}]`; + const row = requireJsonObject(value, label); + const readStatus = requireStringLiteral(row.read_status, SOURCE_READ_STATUSES, `${label}.read_status`); + const lastReadAt = optionalTimestamp(row.last_read_at, `${label}.last_read_at`); + if (readStatus === "read" && lastReadAt === null) { + throw new EffectRuntimeRequestError(`${label}.last_read_at is required for a read source`); + } + if (readStatus === "not_read" && lastReadAt !== null) { + throw new EffectRuntimeRequestError(`${label} cannot be not_read with a last_read_at`); + } + const window = row.window_seconds === undefined || row.window_seconds === null + ? DEFAULT_SOURCE_WINDOW_SECONDS + : count(row.window_seconds, `${label}.window_seconds`); + if (window === 0) throw new EffectRuntimeRequestError(`${label}.window_seconds must be positive`); + return { + source_id: identifier(row.source_id, `${label}.source_id`), + via: row.via === undefined || row.via === null ? null : identifier(row.via, `${label}.via`), + required: row.required === undefined ? true : requireBoolean(row.required, `${label}.required`), + read_status: readStatus, + last_read_at: lastReadAt, + source_updated_at: optionalTimestamp(row.source_updated_at, `${label}.source_updated_at`), + window_seconds: window, + item_count: optionalCount(row.item_count, `${label}.item_count`), + missing_count: optionalCount(row.missing_count, `${label}.missing_count`) ?? 0, + unreadable_count: optionalCount(row.unreadable_count, `${label}.unreadable_count`) ?? 0, + }; +} + +function decodeOmission(value: unknown, index: number): Omission { + const label = `coverage.omitted[${index}]`; + const row = requireJsonObject(value, label); + const reason = requireNonEmptyString(row.reason, `${label}.reason`); + if (!REASON.test(reason)) throw new EffectRuntimeRequestError(`${label}.reason is not a reason code`); + const refs = row.refs === undefined ? [] : boundedArray(row.refs, MAX_REFS, `${label}.refs`).map((ref, refIndex) => { + const text = requireNonEmptyString(ref, `${label}.refs[${refIndex}]`); + if (text.length > MAX_REF_CHARS) throw new EffectRuntimeRequestError(`${label}.refs[${refIndex}] is too long`); + return text; + }); + const omittedCount = count(row.count, `${label}.count`); + if (omittedCount === 0) throw new EffectRuntimeRequestError(`${label}.count must be positive`); + return { reason, count: omittedCount, refs }; +} + +function decodeCoverage(value: unknown): CoverageFact { + const coverage = requireJsonObject(value, "coverage"); + return { + scope: identifier(coverage.scope, "coverage.scope"), + expected_count: optionalCount(coverage.expected_count, "coverage.expected_count"), + included_count: count(coverage.included_count, "coverage.included_count"), + omitted: coverage.omitted === undefined + ? [] + : boundedArray(coverage.omitted, MAX_OMISSIONS, "coverage.omitted").map(decodeOmission), + shown_count: optionalCount(coverage.shown_count, "coverage.shown_count"), + available_count: optionalCount(coverage.available_count, "coverage.available_count"), + }; +} + +function decodeUpstreamSummary(value: unknown, index: number): UpstreamSummary { + const label = `upstream[${index}]`; + const row = requireJsonObject(value, label); + return { + projection: identifier(row.projection, `${label}.projection`), + observed_at: timestamp(row.observed_at, `${label}.observed_at`).text, + complete: requireBoolean(row.complete, `${label}.complete`), + }; +} + +function sealSource(fact: SourceFact, servedMillis: number): JsonObject { + const staleness = fact.last_read_at === null + ? null + : Math.max(0, Math.floor((servedMillis - timestamp(fact.last_read_at, "last_read_at").millis) / 1000)); + let status: SourceStatus; + if (fact.read_status === "read") { + status = staleness !== null && staleness > fact.window_seconds ? "stale" : "fresh"; + } else { + status = fact.read_status; + } + const reasons: string[] = []; + if (status === "stale" || status === "unreadable") reasons.push(status); + if ((status === "missing" || status === "not_read") && fact.required) reasons.push(status); + if (fact.unreadable_count > 0 && status !== "unreadable") reasons.push("partially_unreadable"); + return { + source_id: fact.source_id, + ...(fact.via === null ? {} : { via: fact.via }), + required: fact.required, + read_status: fact.read_status, + last_read_at: fact.last_read_at, + source_updated_at: fact.source_updated_at, + window_seconds: fact.window_seconds, + staleness_seconds: staleness, + status, + item_count: fact.item_count, + missing_count: fact.missing_count, + unreadable_count: fact.unreadable_count, + alert: reasons.length > 0, + alert_reasons: reasons, + }; +} + +function sealFacts( + projection: string, + observed: { text: string; millis: number }, + served: { text: string; millis: number }, + servedFromCache: boolean, + facts: SourceFact[], + coverage: CoverageFact, + upstream: UpstreamSummary[], +): JsonObject { + const seen = new Set(); + for (const fact of facts) { + const key = `${fact.via ?? ""}/${fact.source_id}`; + if (seen.has(key)) throw new EffectRuntimeRequestError(`duplicate source ${key}`); + seen.add(key); + } + const sources = facts.map((fact) => sealSource(fact, served.millis)); + const truncated = coverage.shown_count !== null && coverage.available_count !== null + && coverage.available_count > coverage.shown_count; + const complete = coverage.omitted.length === 0 + && (coverage.expected_count === null || coverage.included_count >= coverage.expected_count) + && upstream.every((row) => row.complete); + const reasons = new Set(); + const alertSourceIds: string[] = []; + for (const row of sources) { + const rowReasons = row.alert_reasons as string[]; + if (rowReasons.length === 0) continue; + alertSourceIds.push(typeof row.via === "string" ? `${row.via}/${row.source_id}` : String(row.source_id)); + for (const reason of rowReasons) { + reasons.add(reason === "partially_unreadable" ? "unreadable_sources" + : reason === "stale" ? "stale_sources" + : reason === "unreadable" ? "unreadable_sources" : "missing_required_sources"); + } + } + if (!complete) reasons.add("incomplete_coverage"); + const fresh = !reasons.has("stale_sources") && !reasons.has("unreadable_sources") + && !reasons.has("missing_required_sources"); + return { + schema_version: PROJECTION_ENVELOPE_SCHEMA_VERSION, + projection, + observed_at: observed.text, + served_at: served.text, + age_seconds: Math.max(0, Math.floor((served.millis - observed.millis) / 1000)), + served_from_cache: servedFromCache, + fresh, + complete, + alert: reasons.size > 0, + alert_reasons: [...reasons].sort(), + alert_source_ids: alertSourceIds, + sources, + coverage: { + scope: coverage.scope, + expected_count: coverage.expected_count, + included_count: coverage.included_count, + omitted: coverage.omitted.map((row) => ({ ...row, refs: [...row.refs] })), + shown_count: coverage.shown_count, + available_count: coverage.available_count, + truncated, + complete, + }, + upstream: upstream.map((row) => ({ ...row })), + }; +} + +function decodeSealedEnvelope(value: unknown, label: string): { + projection: string; + observed: { text: string; millis: number }; + sources: SourceFact[]; + coverage: CoverageFact; + upstream: UpstreamSummary[]; + complete: boolean; +} { + const envelope = requireJsonObject(value, label); + if (envelope.schema_version !== PROJECTION_ENVELOPE_SCHEMA_VERSION) { + throw new EffectRuntimeRequestError(`${label} schema mismatch`); + } + const sources = boundedArray(envelope.sources, MAX_SOURCES, `${label}.sources`).map(decodeSource); + const coverage = decodeCoverage(envelope.coverage); + const upstream = boundedArray(envelope.upstream ?? [], MAX_UPSTREAM, `${label}.upstream`) + .map(decodeUpstreamSummary); + const projection = identifier(envelope.projection, `${label}.projection`); + const observed = timestamp(envelope.observed_at, `${label}.observed_at`); + const complete = coverage.omitted.length === 0 + && (coverage.expected_count === null || coverage.included_count >= coverage.expected_count) + && upstream.every((row) => row.complete); + return { projection, observed, sources, coverage, upstream, complete }; +} + +/** Seal compact read facts, optionally composing upstream sealed envelopes. A + * derived projection inherits every upstream source row, so its staleness is + * bounded by the oldest read it depends on, never by its own assembly time. */ +export function sealProjectionEnvelope(value: unknown): JsonObject { + const request = requireJsonObject(value, "projection envelope request"); + if (request.schema_version === PROJECTION_ENVELOPE_SERVE_REQUEST) { + const sealed = decodeSealedEnvelope(request.envelope, "envelope"); + return sealFacts( + sealed.projection, + sealed.observed, + timestamp(request.served_at, "served_at"), + true, + sealed.sources, + sealed.coverage, + sealed.upstream, + ); + } + if (request.schema_version !== PROJECTION_ENVELOPE_SEAL_REQUEST) { + throw new EffectRuntimeRequestError("projection envelope request schema mismatch"); + } + const projection = identifier(request.projection, "projection"); + const observed = timestamp(request.observed_at, "observed_at"); + const served = request.served_at === undefined || request.served_at === null + ? observed + : timestamp(request.served_at, "served_at"); + const facts = boundedArray(request.sources, MAX_SOURCES, "sources").map(decodeSource); + const upstream: UpstreamSummary[] = []; + const upstreamValues = boundedArray(request.upstream ?? [], MAX_UPSTREAM, "upstream"); + upstreamValues.forEach((value, index) => { + const sealed = decodeSealedEnvelope(value, `upstream[${index}]`); + upstream.push({ projection: sealed.projection, observed_at: sealed.observed.text, complete: sealed.complete }); + for (const fact of sealed.sources) facts.push({ ...fact, via: fact.via ?? sealed.projection }); + }); + if (facts.length > MAX_SOURCES) throw new EffectRuntimeRequestError(`sources exceed ${MAX_SOURCES} entries`); + return sealFacts(projection, observed, served, false, facts, decodeCoverage(request.coverage), upstream); +} diff --git a/loopx/control_plane/projection_envelope_facts.py b/loopx/control_plane/projection_envelope_facts.py new file mode 100644 index 0000000000..8c908a0a5e --- /dev/null +++ b/loopx/control_plane/projection_envelope_facts.py @@ -0,0 +1,115 @@ +"""Python facts adapter for the TypeScript-owned projection envelope. + +Python-owned projections record when each source was read and what scope they +covered; `projection_envelope.ts` alone decides freshness, alerts and +completeness. The adapter exits when the projection itself migrates. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any + +from .effect_runtime import effect_runtime_result +from .runtime.time import now_utc_iso + +PROJECTION_ENVELOPE_SCHEMA_VERSION = "loopx_projection_envelope_v0" +SEAL_REQUEST_SCHEMA_VERSION = "loopx_projection_envelope_seal_request_v0" +SERVE_REQUEST_SCHEMA_VERSION = "loopx_projection_envelope_serve_request_v0" +ALERT_MARKER = "🔴" + + +def source_fact( + source_id: str, + *, + read_status: str = "read", + last_read_at: str | None = None, + required: bool = True, + source_updated_at: str | None = None, + item_count: int | None = None, + missing_count: int = 0, + unreadable_count: int = 0, +) -> dict[str, Any]: + return { + "source_id": source_id, + "read_status": read_status, + "last_read_at": last_read_at, + "required": required, + "source_updated_at": source_updated_at, + "item_count": item_count, + "missing_count": missing_count, + "unreadable_count": unreadable_count, + } + + +def seal_projection_envelope( + *, + projection: str, + observed_at: str, + sources: Sequence[Mapping[str, Any]], + coverage: Mapping[str, Any], + upstream: Sequence[Mapping[str, Any]] = (), +) -> dict[str, Any]: + return effect_runtime_result( + "projection.envelope.seal", + { + "schema_version": SEAL_REQUEST_SCHEMA_VERSION, + "projection": projection, + "observed_at": observed_at, + "sources": [dict(row) for row in sources], + "coverage": dict(coverage), + "upstream": [dict(row) for row in upstream], + }, + ) + + +def serve_projection_envelope( + envelope: Mapping[str, Any], *, served_at: str | None = None +) -> dict[str, Any]: + """Re-serve a stored envelope; staleness is recomputed, observed_at kept.""" + + return effect_runtime_result( + "projection.envelope.seal", + { + "schema_version": SERVE_REQUEST_SCHEMA_VERSION, + "envelope": dict(envelope), + "served_at": served_at or now_utc_iso(), + }, + ) + + +def render_projection_envelope_markdown(envelope: Any) -> list[str]: + if not isinstance(envelope, Mapping): + return [f"- projection: {ALERT_MARKER} no projection envelope; freshness and coverage unknown"] + coverage = envelope.get("coverage") if isinstance(envelope.get("coverage"), Mapping) else {} + sources = [row for row in envelope.get("sources") or [] if isinstance(row, Mapping)] + absent_optional = sum( + 1 for row in sources if row.get("required") is False and row.get("read_status") != "read" + ) + fresh_count = sum(1 for row in sources if row.get("status") == "fresh") + marker = f"{ALERT_MARKER} " if envelope.get("alert") else "" + line = ( + f"- projection: {marker}observed_at=`{envelope.get('observed_at')}` " + f"age=`{envelope.get('age_seconds')}s`" + + (" (cached)" if envelope.get("served_from_cache") else "") + + f" sources_fresh=`{fresh_count}/{len(sources) - absent_optional}`" + + (f" (+{absent_optional} optional absent)" if absent_optional else "") + + f" coverage=`{coverage.get('included_count')}/{coverage.get('expected_count')}` " + f"scope=`{coverage.get('scope')}`" + ) + if coverage.get("truncated"): + line += f" shown=`{coverage.get('shown_count')}/{coverage.get('available_count')}`" + lines = [line] + if envelope.get("alert"): + alerts = ",".join(envelope.get("alert_reasons") or []) + ids = ",".join(envelope.get("alert_source_ids") or []) or "-" + omitted = ",".join( + f"{row.get('reason')}={row.get('count')}" + for row in coverage.get("omitted") or [] + if isinstance(row, Mapping) + ) or "-" + lines.append( + f"- {ALERT_MARKER} projection alerts: `{alerts}` sources=`{ids}` omitted=`{omitted}`; " + "state this before treating the projection as current or whole" + ) + return lines diff --git a/loopx/control_plane/runtime/status_projection_cache.py b/loopx/control_plane/runtime/status_projection_cache.py index 77dc52d103..5555bf847d 100644 --- a/loopx/control_plane/runtime/status_projection_cache.py +++ b/loopx/control_plane/runtime/status_projection_cache.py @@ -10,6 +10,8 @@ from ...history import load_registry from ...paths import resolve_runtime_root +from ..effect_runtime import EffectRuntimeRejected +from ..projection_envelope_facts import serve_projection_envelope from ..todos.contract import normalize_required_capabilities from .time import now_utc as runtime_now_utc from .time import now_utc_iso as runtime_now_utc_iso @@ -222,9 +224,20 @@ def load_status_projection_cache( if not isinstance(payload, dict): metadata["miss_reason"] = "missing_payload" return None, metadata + envelope = payload.get("projection_envelope") + if not isinstance(envelope, dict): + metadata["miss_reason"] = "missing_projection_envelope" + return None, metadata + try: + served_envelope = serve_projection_envelope(envelope) + except EffectRuntimeRejected as exc: + metadata["miss_reason"] = "invalid_projection_envelope" + metadata["error"] = str(exc) + return None, metadata metadata["hit"] = True metadata["miss_reason"] = None payload = dict(payload) + payload["projection_envelope"] = served_envelope payload["projection_cache"] = dict(metadata) return payload, metadata diff --git a/loopx/control_plane/status/collection.py b/loopx/control_plane/status/collection.py index fa318bc6f6..35d695eddb 100644 --- a/loopx/control_plane/status/collection.py +++ b/loopx/control_plane/status/collection.py @@ -19,6 +19,8 @@ from ..runtime.runtime_projection_route import ( collect_runtime_projection_route_diagnostics, ) +from ..projection_envelope_facts import seal_projection_envelope, source_fact +from ..runtime.time import now_utc_iso, parse_timestamp, utc_isoformat from ..todos.todo_index import MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL from ...registry import registry_goals from ...rollout_event_log import RolloutEventSnapshot @@ -93,6 +95,7 @@ def collect_status( else None ) registry = context.load_registry(registry_path) + registry_read_at = now_utc_iso() runtime_root = context.resolve_runtime_root( registry, runtime_root_override, @@ -107,6 +110,7 @@ def collect_status( runtime_root=runtime_root, current_registry=registry, ) + global_registry_read_at = now_utc_iso() include_runtime_goals = bool(global_registry.get("current_registry_is_global")) history_collection = context.collect_status_history( registry_path=registry_path, @@ -119,6 +123,7 @@ def collect_status( registry=registry, ) history = history_collection.status_history + history_read_at = now_utc_iso() contract = context.check_contract( registry_path=registry_path, runtime_root_override=str(runtime_root), @@ -130,6 +135,7 @@ def collect_status( history_audit=history_collection.contract_audit, registry=registry, ) + contract_read_at = now_utc_iso() contract = project_contract_health_for_goal(contract, goal_id=goal_filter) queue = context.build_attention_queue( contract=contract, @@ -168,6 +174,7 @@ def collect_status( activation_state_filter=activation_filter, registry=registry, ) + routes_read_at = now_utc_iso() runtime_projection_route_health = { "healthy": ( bool(runtime_projection_routes.get("healthy")) @@ -263,4 +270,97 @@ def collect_status( ) attach_goal_acceptance_observations(payload, history=history) attach_goal_artifact_lifecycle_projections(payload, history=history) + payload["projection_envelope"] = seal_projection_envelope( + projection="status", + observed_at=now_utc_iso(), + sources=[ + source_fact("registry", last_read_at=registry_read_at, item_count=len(registry_goals(registry))), + source_fact( + "global_registry", + read_status="read" if global_registry.get("available") else "missing", + last_read_at=global_registry_read_at if global_registry.get("available") else None, + required=False, + item_count=global_registry.get("global_goal_count"), + ), + _run_index_source(history, read_at=history_read_at), + source_fact("goal_state_contract", last_read_at=contract_read_at), + source_fact( + "runtime_projection_routes", + read_status="read" if runtime_projection_routes.get("available") else "missing", + last_read_at=routes_read_at if runtime_projection_routes.get("available") else None, + required=False, + item_count=runtime_projection_route_health["goal_count"], + ), + ], + coverage=_status_coverage( + registry, + history=history, + queue=queue, + goal_filter=goal_filter, + activation_filter=activation_filter, + ), + ) return payload + + +def _run_index_source(history: dict[str, Any], *, read_at: str) -> dict[str, Any]: + goals = [goal for goal in history.get("goals") or [] if isinstance(goal, dict)] + newest = None + for goal in goals: + for run in goal.get("latest_runs") or []: + generated = parse_timestamp(run.get("generated_at")) if isinstance(run, dict) else None + if generated is not None and generated.tzinfo is not None and (newest is None or generated > newest): + newest = generated + return source_fact( + "goal_run_indexes", + last_read_at=read_at, + source_updated_at=utc_isoformat(newest) if newest is not None else None, + item_count=len(goals), + missing_count=sum(1 for goal in goals if goal.get("index_exists") is False), + ) + + +def _status_coverage( + registry: dict[str, Any], + *, + history: dict[str, Any], + queue: dict[str, Any], + goal_filter: str | None, + activation_filter: GoalActivationState | None, +) -> dict[str, Any]: + """Count registry members in the requested scope; legacy runtime goals are extra.""" + members = registry_goals(registry) + if goal_filter is not None: + scope, expected = "goal", 1 + expected_ids = {goal_filter} + elif activation_filter is not None: + scope = f"activation.{activation_filter.value}" + expected_ids = { + str(goal.get("id") or "") for goal in members if goal_activation_state(goal) is activation_filter + } + expected = len(expected_ids) + else: + scope = "registry" + expected_ids = {str(goal.get("id") or "") for goal in members} + expected = len(expected_ids) + projected = { + str(goal.get("id") or "") + for goal in history.get("goals") or [] + if isinstance(goal, dict) + and (goal.get("registry_member") is True or (goal_filter is not None and goal.get("index_exists") is True)) + } + missing = sorted(expected_ids - projected) + items = queue.get("items") if isinstance(queue.get("items"), list) else [] + item_count = queue.get("item_count") + return { + "scope": scope, + "expected_count": expected, + "included_count": len(expected_ids & projected), + "omitted": ( + [{"reason": "goal_not_found" if goal_filter else "not_projected", "count": len(missing), "refs": missing[:8]}] + if missing + else [] + ), + "shown_count": len(items), + "available_count": item_count if isinstance(item_count, int) and item_count >= 0 else len(items), + } diff --git a/loopx/presentation/renderers/status_markdown.py b/loopx/presentation/renderers/status_markdown.py index c9f1395165..311ce5e402 100644 --- a/loopx/presentation/renderers/status_markdown.py +++ b/loopx/presentation/renderers/status_markdown.py @@ -4,6 +4,7 @@ from typing import Any from ...control_plane import control_plane_policy_summary +from ...control_plane.projection_envelope_facts import render_projection_envelope_markdown from ...control_plane.runtime.event_ledger import EVENT_LEDGER_CLASSES from ...execution_profile import execution_profile_summary from ...long_task_cadence import long_task_cadence_hint_summary @@ -61,6 +62,7 @@ def append_status_overview_markdown( f"- runtime_root: `{payload.get('runtime_root')}`", f"- goals: `{payload.get('goal_count')}`", f"- runs: `{payload.get('run_count')}`", + *render_projection_envelope_markdown(payload.get("projection_envelope")), ] ) diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index 62e987d6e8..c8b4f7eb37 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -1375,7 +1375,7 @@ }, { "site": "loopx/control_plane/runtime/status_projection_cache.py::.resolve_status_projection_cache_runtime_root::codec_read:load_registry#1", - "line": 39, + "line": 41, "column": 16, "kind": "codec_read", "api": "load_registry", @@ -1383,7 +1383,7 @@ }, { "site": "loopx/control_plane/status/collection.py::.collect_status::codec_read:load_registry#1", - "line": 95, + "line": 97, "column": 16, "kind": "codec_read", "api": "load_registry", diff --git a/loopx/summary_all.py b/loopx/summary_all.py index a2c5ed6d90..e1e16ae754 100644 --- a/loopx/summary_all.py +++ b/loopx/summary_all.py @@ -4,6 +4,11 @@ from pathlib import Path from typing import Any +from .control_plane.projection_envelope_facts import ( + render_projection_envelope_markdown, + seal_projection_envelope, + source_fact, +) from .control_plane.runtime.time import now_utc, now_utc_iso from .control_plane.todos.decision_scope import todo_gate_relations from .control_plane.todos.user_gate import open_user_gate_todo_items @@ -128,6 +133,60 @@ def _build_goal_quota( } +def _quota_unavailable(quota_payload: dict[str, Any]) -> bool: + return quota_payload.get("state") == "quota_unavailable" + + +def _global_projection_envelope( + projection: str, + status_payload: dict[str, Any], + *, + quota_read_at: str, + quota_evaluated: int, + quota_unavailable: int, + shown_count: int, + available_count: int, +) -> dict[str, Any]: + """A global read is complete only when it covers every global-registry goal.""" + global_registry = _as_dict(status_payload.get("global_registry")) + status_envelope = _as_dict(status_payload.get("projection_envelope")) + status_coverage = _as_dict(status_envelope.get("coverage")) + excluded = 0 + refs: list[str] = [] + if global_registry.get("available"): + expected = int(global_registry.get("global_goal_count") or 0) + excluded = int(global_registry.get("current_registry_excluded_goal_count") or 0) + refs = [str(ref) for ref in _as_list(global_registry.get("current_registry_excluded_goal_ids"))] + else: + expected = int(status_coverage.get("expected_count") or 0) + return seal_projection_envelope( + projection=projection, + observed_at=_now_iso(), + sources=[ + source_fact( + "goal_quota", + last_read_at=quota_read_at, + item_count=quota_evaluated, + unreadable_count=quota_unavailable, + ), + *([] if status_envelope else [source_fact("status", read_status="not_read")]), + ], + coverage={ + "scope": "global", + "expected_count": expected, + "included_count": max(0, expected - excluded), + "omitted": ( + [{"reason": "outside_current_registry", "count": excluded, "refs": refs[:8]}] + if excluded + else [] + ), + "shown_count": shown_count, + "available_count": available_count, + }, + upstream=[status_envelope] if status_envelope else [], + ) + + def _lane_from_item( item: dict[str, Any], *, @@ -315,7 +374,7 @@ def _collect_global_gate_state( *, agent_id: str | None, limit: int, -) -> dict[str, list[dict[str, Any]]]: +) -> dict[str, Any]: queue = _as_dict(status_payload.get("attention_queue")) queue_items = [ item for item in _as_list(queue.get("items")) if isinstance(item, dict) @@ -323,6 +382,7 @@ def _collect_global_gate_state( gates: list[dict[str, Any]] = [] lanes: list[dict[str, Any]] = [] seen_goal_ids: set[str] = set() + quota_evaluated = quota_unavailable = 0 for item in queue_items: goal_id = str(item.get("goal_id") or "").strip() if not goal_id or goal_id in seen_goal_ids: @@ -333,6 +393,8 @@ def _collect_global_gate_state( goal_id=goal_id, agent_id=agent_id, ) + quota_evaluated += 1 + quota_unavailable += int(_quota_unavailable(quota_payload)) if not _quota_matches_agent_scope(quota_payload, agent_id=agent_id): continue gate = _global_gate_from_item( @@ -354,7 +416,15 @@ def _collect_global_gate_state( lanes.append(lane) if len(gates) >= limit: break - return {"gates": gates, "lanes": lanes} + unique_goal_count = len({str(item.get("goal_id") or "").strip() for item in queue_items} - {""}) + return { + "gates": gates, + "lanes": lanes, + "quota_evaluated": quota_evaluated, + "quota_unavailable": quota_unavailable, + "scanned_goal_count": len(seen_goal_ids), + "queue_goal_count": unique_goal_count, + } def _global_gates_request() -> dict[str, Any]: @@ -386,11 +456,21 @@ def build_global_gates( ) if status_payload.get("ok") is not True: return build_global_gates_error("Global status source unavailable.") + quota_read_at = _now_iso() state = _collect_global_gate_state( status_payload, agent_id=agent_id, limit=normalized_limit, ) + envelope = _global_projection_envelope( + "global_gates", + status_payload, + quota_read_at=quota_read_at, + quota_evaluated=state["quota_evaluated"], + quota_unavailable=state["quota_unavailable"], + shown_count=state["scanned_goal_count"], + available_count=state["queue_goal_count"], + ) gates = state["gates"] lanes = state["lanes"] gate_count = len(gates) @@ -444,6 +524,7 @@ def build_global_gates( ), "Recent progress and unrelated risks are outside this focused command.", ], + "projection_envelope": envelope, "boundary": BOUNDARY, } @@ -474,6 +555,7 @@ def render_global_gates_markdown(payload: dict[str, Any]) -> str: "", f"- command: `{_as_dict(payload.get('request')).get('command')}`", f"- open_gate_count: `{summary.get('open_gate_count')}`", + *render_projection_envelope_markdown(payload.get("projection_envelope")), "", "## Gates", ] @@ -576,6 +658,8 @@ def build_summary_all( todos: list[dict[str, Any]] = [] quota_states: dict[str, int] = {} seen_goal_ids: set[str] = set() + quota_unavailable = 0 + quota_read_at = _now_iso() for item in queue_items[: max(limit * 4, 40)]: goal_id = str(item.get("goal_id") or "").strip() @@ -583,6 +667,7 @@ def build_summary_all( continue seen_goal_ids.add(goal_id) quota_payload = _build_goal_quota(status_payload, goal_id=goal_id, agent_id=agent_id) + quota_unavailable += int(_quota_unavailable(quota_payload)) lane = _lane_from_item(item, quota_payload=quota_payload) if lane: lanes.append(lane) @@ -594,6 +679,15 @@ def build_summary_all( if todo: todos.append(todo) lanes.sort(key=_lane_sort_key) + envelope = _global_projection_envelope( + "global_summary", + status_payload, + quota_read_at=quota_read_at, + quota_evaluated=len(seen_goal_ids), + quota_unavailable=quota_unavailable, + shown_count=min(len(lanes), limit), + available_count=len({str(item.get("goal_id") or "").strip() for item in queue_items} - {""}), + ) global_registry = _as_dict(status_payload.get("global_registry")) risks = [_risk_from_finding(item) for item in _as_list(global_registry.get("findings")) if isinstance(item, dict)] @@ -673,6 +767,7 @@ def build_summary_all( "Raw logs, raw transcripts, connector payloads, credential values, local paths, and private source bodies were intentionally omitted.", "Status health findings are summarized without filesystem paths.", ], + "projection_envelope": envelope, "boundary": BOUNDARY, } return payload @@ -690,6 +785,7 @@ def render_summary_all_markdown(payload: dict[str, Any]) -> str: f"- time_range: `{_as_dict(payload.get('request')).get('time_range')}`", f"- headline: {summary.get('headline')}", f"- counts: progress=`{summary.get('progress_count')}`, gates=`{summary.get('open_gate_count')}`, todos=`{summary.get('runnable_todo_count')}`, risks=`{summary.get('risk_count')}`", + *render_projection_envelope_markdown(payload.get("projection_envelope")), "", "## Lanes", ] diff --git a/tests/control_plane_ts/projection_envelope.test.ts b/tests/control_plane_ts/projection_envelope.test.ts new file mode 100644 index 0000000000..871faa27c6 --- /dev/null +++ b/tests/control_plane_ts/projection_envelope.test.ts @@ -0,0 +1,182 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + DEFAULT_SOURCE_WINDOW_SECONDS, + PROJECTION_ENVELOPE_SCHEMA_VERSION, + PROJECTION_ENVELOPE_SEAL_REQUEST, + PROJECTION_ENVELOPE_SERVE_REQUEST, + sealProjectionEnvelope, +} from "../../loopx/control_plane/projection_envelope.ts"; +import { EffectRuntimeRequestError } from "../../loopx/control_plane/effect_runtime_errors.ts"; + +const OBSERVED = "2026-09-26T10:00:00+00:00"; +const at = (seconds: number) => new Date(Date.parse(OBSERVED) + seconds * 1000).toISOString(); + +function sealRequest(overrides: Record = {}) { + return { + schema_version: PROJECTION_ENVELOPE_SEAL_REQUEST, + projection: "status", + observed_at: OBSERVED, + sources: [ + { source_id: "registry", read_status: "read", last_read_at: OBSERVED, item_count: 3 }, + { source_id: "global_registry", read_status: "missing", required: false }, + ], + coverage: { scope: "registry", expected_count: 3, included_count: 3 }, + ...overrides, + }; +} + +type Row = Record; +const sources = (envelope: Row) => envelope.sources as Row[]; +const coverage = (envelope: Row) => envelope.coverage as Row; + +test("a live read inside its window is fresh and complete", () => { + const envelope = sealProjectionEnvelope(sealRequest()); + assert.equal(envelope.schema_version, PROJECTION_ENVELOPE_SCHEMA_VERSION); + assert.equal(envelope.served_at, OBSERVED); + assert.equal(envelope.age_seconds, 0); + assert.equal(envelope.served_from_cache, false); + assert.equal(envelope.fresh, true); + assert.equal(envelope.complete, true); + assert.equal(envelope.alert, false); + assert.deepEqual(envelope.alert_reasons, []); + const [registry, global] = sources(envelope); + assert.equal(registry.status, "fresh"); + assert.equal(registry.staleness_seconds, 0); + assert.equal(registry.window_seconds, DEFAULT_SOURCE_WINDOW_SECONDS); + assert.equal(global.status, "missing"); + assert.equal(global.alert, false, "an optional missing source is disclosed but not alerted"); +}); + +test("staleness is measured at serve time against each source window", () => { + const envelope = sealProjectionEnvelope(sealRequest({ + served_at: at(90), + sources: [ + { source_id: "registry", read_status: "read", last_read_at: at(-400) }, + { source_id: "goal_run_indexes", read_status: "read", last_read_at: OBSERVED, window_seconds: 60 }, + { source_id: "contract", read_status: "read", last_read_at: OBSERVED, window_seconds: 120 }, + ], + })); + assert.deepEqual(sources(envelope).map((row) => [row.source_id, row.status, row.staleness_seconds]), [ + ["registry", "stale", 490], + ["goal_run_indexes", "stale", 90], + ["contract", "fresh", 90], + ]); + assert.equal(envelope.fresh, false); + assert.deepEqual(envelope.alert_reasons, ["stale_sources"]); + assert.deepEqual(envelope.alert_source_ids, ["registry", "goal_run_indexes"]); +}); + +test("unreadable, partial and required-but-missing sources alert", () => { + const envelope = sealProjectionEnvelope(sealRequest({ + sources: [ + { source_id: "registry", read_status: "unreadable" }, + { source_id: "goal_quota", read_status: "read", last_read_at: OBSERVED, item_count: 4, unreadable_count: 1 }, + { source_id: "global_registry", read_status: "missing", required: true }, + { source_id: "rollout_events", read_status: "not_read", required: false }, + ], + })); + assert.deepEqual(sources(envelope).map((row) => row.alert_reasons), [ + ["unreadable"], ["partially_unreadable"], ["missing"], [], + ]); + assert.deepEqual(envelope.alert_reasons, ["missing_required_sources", "unreadable_sources"]); + assert.equal(envelope.fresh, false); +}); + +test("coverage below the requested scope is incomplete; display truncation is disclosed separately", () => { + const envelope = sealProjectionEnvelope(sealRequest({ + coverage: { + scope: "global", + expected_count: 12, + included_count: 9, + omitted: [{ reason: "outside_current_registry", count: 3, refs: ["a", "b", "c"] }], + shown_count: 5, + available_count: 9, + }, + })); + assert.equal(envelope.complete, false); + assert.equal(coverage(envelope).truncated, true); + assert.deepEqual(envelope.alert_reasons, ["incomplete_coverage"]); + + const truncatedOnly = sealProjectionEnvelope(sealRequest({ + coverage: { scope: "registry", expected_count: 3, included_count: 3, shown_count: 1, available_count: 3 }, + })); + assert.equal(coverage(truncatedOnly).truncated, true); + assert.equal(truncatedOnly.complete, true); + assert.equal(truncatedOnly.alert, false, "a requested display limit is not an alert"); + + const undercounted = sealProjectionEnvelope(sealRequest({ + coverage: { scope: "goal", expected_count: 1, included_count: 0 }, + })); + assert.equal(undercounted.complete, false, "a shortfall without a named omission is still incomplete"); +}); + +test("serving a cached copy keeps observed_at and recomputes staleness", () => { + const sealed = sealProjectionEnvelope(sealRequest()); + const served = sealProjectionEnvelope({ + schema_version: PROJECTION_ENVELOPE_SERVE_REQUEST, + envelope: JSON.parse(JSON.stringify(sealed)), + served_at: at(600), + }); + assert.equal(served.observed_at, OBSERVED); + assert.equal(served.served_at, at(600)); + assert.equal(served.age_seconds, 600); + assert.equal(served.served_from_cache, true); + assert.equal(sources(served)[0].status, "stale"); + assert.deepEqual(served.alert_reasons, ["stale_sources"]); + assert.deepEqual(coverage(served), coverage(sealed)); +}); + +test("a derived projection inherits upstream sources and completeness", () => { + const status = sealProjectionEnvelope(sealRequest({ + sources: [{ source_id: "registry", read_status: "read", last_read_at: at(-500) }], + served_at: OBSERVED, + coverage: { scope: "registry", expected_count: 3, included_count: 2 }, + })); + const summary = sealProjectionEnvelope({ + schema_version: PROJECTION_ENVELOPE_SEAL_REQUEST, + projection: "global_summary", + observed_at: at(1), + sources: [{ source_id: "goal_quota", read_status: "read", last_read_at: at(1), item_count: 2 }], + coverage: { scope: "global", expected_count: 2, included_count: 2 }, + upstream: [status], + }); + assert.deepEqual(sources(summary).map((row) => [row.via ?? null, row.source_id, row.status]), [ + [null, "goal_quota", "fresh"], + ["status", "registry", "stale"], + ]); + assert.deepEqual(summary.upstream, [{ projection: "status", observed_at: OBSERVED, complete: false }]); + assert.equal(summary.complete, false, "an incomplete upstream cannot yield a complete derived projection"); + assert.deepEqual(summary.alert_source_ids, ["status/registry"]); + + const reserved = sealProjectionEnvelope({ + schema_version: PROJECTION_ENVELOPE_SERVE_REQUEST, envelope: summary, served_at: at(2), + }); + assert.deepEqual(reserved.upstream, summary.upstream); + assert.equal(sources(reserved)[1].via, "status"); +}); + +test("the decoder rejects values that cannot establish freshness", () => { + const rejects = (overrides: Record, pattern: RegExp) => + assert.throws(() => sealProjectionEnvelope(sealRequest(overrides)), (error: unknown) => + error instanceof EffectRuntimeRequestError && pattern.test(error.message)); + rejects({ schema_version: "other" }, /schema mismatch/); + rejects({ observed_at: "2026-09-26T10:00:00" }, /explicit offset/); + rejects({ observed_at: "2026-02-30T10:00:00Z" }, /explicit offset/); + rejects({ sources: [{ source_id: "registry", read_status: "read" }] }, /required for a read source/); + rejects({ sources: [{ source_id: "registry", read_status: "not_read", last_read_at: OBSERVED }] }, /not_read/); + rejects({ sources: [{ source_id: "Registry Path", read_status: "missing" }] }, /identifier/); + rejects({ sources: [{ source_id: "registry", read_status: "fresh" }] }, /unsupported/); + rejects({ sources: [ + { source_id: "registry", read_status: "missing" }, { source_id: "registry", read_status: "missing" }, + ] }, /duplicate source/); + rejects({ sources: [{ source_id: "registry", read_status: "missing", window_seconds: 0 }] }, /positive/); + rejects({ coverage: { scope: "registry", included_count: -1 } }, /non-negative/); + rejects({ coverage: { scope: "registry", included_count: 1, omitted: [{ reason: "x", count: 0 }] } }, /positive/); + rejects({ sources: Array.from({ length: 33 }, (_, index) => ({ source_id: `s${index}`, read_status: "missing" })) }, + /exceeds 32/); + assert.throws(() => sealProjectionEnvelope({ + schema_version: PROJECTION_ENVELOPE_SERVE_REQUEST, envelope: { schema_version: "other" }, served_at: OBSERVED, + }), /schema mismatch/); +}); diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py new file mode 100644 index 0000000000..bc215c6ad3 --- /dev/null +++ b/tests/test_projection_envelope.py @@ -0,0 +1,259 @@ +from __future__ import annotations + +import json +from datetime import timedelta +from pathlib import Path + +import loopx.control_plane.projection_envelope_facts as projection_envelope +import loopx.summary_all as summary_all +from loopx.control_plane.runtime.status_projection_cache import ( + load_status_projection_cache, + status_projection_cache_path, + status_projection_cache_key, + write_status_projection_cache, +) +from loopx.control_plane.runtime.time import now_utc, utc_isoformat +from loopx.presentation.renderers.status_markdown import render_status_markdown +from loopx.status import collect_status + + +def _registry(tmp_path: Path, goal_ids: tuple[str, ...] = ("goal-a", "goal-b")) -> Path: + registry_path = tmp_path / ".loopx" / "registry.json" + registry_path.parent.mkdir(parents=True) + (tmp_path / "ACTIVE_GOAL_STATE.md").write_text("# Goal\n", encoding="utf-8") + registry_path.write_text( + json.dumps( + { + "common_runtime_root": str(tmp_path / "runtime"), + "goals": [ + { + "id": goal_id, + "objective": "Project envelope fixture.", + "repo": str(tmp_path), + "state_file": str(tmp_path / "ACTIVE_GOAL_STATE.md"), + "adapter": {"kind": "read_only_project_map_v0"}, + } + for goal_id in goal_ids + ], + } + ), + encoding="utf-8", + ) + return registry_path + + +def _status(registry_path: Path, **kwargs): + return collect_status( + registry_path=registry_path, + runtime_root_override=None, + scan_roots=[], + limit=5, + include_public_boundary_scan=False, + **kwargs, + ) + + +def _sources(envelope: dict) -> dict[str, dict]: + return { + (f"{row['via']}/" if row.get("via") else "") + row["source_id"]: row + for row in envelope["sources"] + } + + +def test_status_envelope_discloses_reads_and_registry_coverage(tmp_path: Path) -> None: + payload = _status(_registry(tmp_path)) + envelope = payload["projection_envelope"] + + assert envelope["schema_version"] == projection_envelope.PROJECTION_ENVELOPE_SCHEMA_VERSION + assert envelope["projection"] == "status" + assert envelope["served_at"] == envelope["observed_at"] + assert envelope["coverage"]["scope"] == "registry" + assert (envelope["coverage"]["expected_count"], envelope["coverage"]["included_count"]) == (2, 2) + assert envelope["complete"] is True and envelope["fresh"] is True and envelope["alert"] is False + sources = _sources(envelope) + assert set(sources) == { + "registry", "global_registry", "goal_run_indexes", "goal_state_contract", "runtime_projection_routes", + } + assert sources["registry"]["item_count"] == 2 + assert sources["global_registry"]["status"] == "missing" + assert sources["global_registry"]["alert"] is False + assert sources["goal_run_indexes"]["missing_count"] == 2 + for row in sources.values(): + assert row["last_read_at"] is None or row["last_read_at"] <= envelope["observed_at"] + assert str(tmp_path) not in json.dumps(envelope), "the envelope must stay path-free" + + +def test_goal_status_for_an_unknown_goal_is_incomplete(tmp_path: Path) -> None: + envelope = _status(_registry(tmp_path), goal_id="goal-missing")["projection_envelope"] + + assert envelope["coverage"]["scope"] == "goal" + 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_goal_status_scopes_coverage_to_the_goal(tmp_path: Path) -> None: + envelope = _status(_registry(tmp_path), goal_id="goal-a")["projection_envelope"] + + assert (envelope["coverage"]["scope"], envelope["coverage"]["expected_count"]) == ("goal", 1) + assert envelope["complete"] is True + + +def _cache_args(registry_path: Path) -> dict: + return { + "registry_path": registry_path, + "runtime_root": registry_path.parents[1] / "runtime", + "scan_roots": [], + "limit": 5, + "include_task_graph": False, + "goal_id": None, + } + + +def test_cached_status_keeps_observed_at_and_restamps_staleness(tmp_path: Path, monkeypatch) -> None: + registry_path = _registry(tmp_path) + payload = _status(registry_path) + write_status_projection_cache(payload=payload, max_age_seconds=3600, **_cache_args(registry_path)) + later = utc_isoformat(now_utc() + timedelta(seconds=900)) + monkeypatch.setattr(projection_envelope, "now_utc_iso", lambda: later) + + cached, metadata = load_status_projection_cache(max_age_seconds=3600, **_cache_args(registry_path)) + + assert metadata["hit"] is True + envelope = cached["projection_envelope"] + assert envelope["observed_at"] == payload["projection_envelope"]["observed_at"] + assert envelope["served_at"] == later + assert envelope["served_from_cache"] is True + assert envelope["age_seconds"] >= 900 + assert "stale_sources" in envelope["alert_reasons"] + assert "(cached)" in render_status_markdown(cached) + + +def test_cache_entry_without_an_envelope_is_a_miss(tmp_path: Path) -> None: + registry_path = _registry(tmp_path) + args = _cache_args(registry_path) + payload = _status(registry_path) + payload.pop("projection_envelope") + write_status_projection_cache(payload=payload, max_age_seconds=3600, **args) + + cached, metadata = load_status_projection_cache(max_age_seconds=3600, **args) + + assert cached is None + assert metadata["miss_reason"] == "missing_projection_envelope" + + +def test_cache_entry_with_a_tampered_envelope_is_a_miss(tmp_path: Path) -> None: + registry_path = _registry(tmp_path) + args = _cache_args(registry_path) + write_status_projection_cache(payload=_status(registry_path), max_age_seconds=3600, **args) + path = status_projection_cache_path( + args["runtime_root"], + status_projection_cache_key( + **{key: value for key, value in args.items()}, + ), + ) + record = json.loads(path.read_text(encoding="utf-8")) + record["payload"]["projection_envelope"]["observed_at"] = "yesterday" + path.write_text(json.dumps(record), encoding="utf-8") + + cached, metadata = load_status_projection_cache(max_age_seconds=3600, **args) + + assert cached is None + assert metadata["miss_reason"] == "invalid_projection_envelope" + + +def _global_status_payload(tmp_path: Path, *, excluded: int) -> dict: + payload = _status(_registry(tmp_path, ("goal-a",))) + payload["ok"] = True + payload["global_registry"] = { + "available": True, + "ok": True, + "current_registry_is_global": excluded == 0, + "global_goal_count": 1 + excluded, + "current_goal_count": 1, + "current_registry_excluded_goal_count": excluded, + "current_registry_excluded_goal_ids": [f"other-{index}" for index in range(min(excluded, 8))], + "findings": [], + } + payload["attention_queue"] = {"items": [{"goal_id": "goal-a", "waiting_on": "codex"}], "item_count": 1} + return payload + + +def _patch_status(monkeypatch, status_payload: dict) -> None: + monkeypatch.setattr(summary_all, "collect_status", lambda **_: status_payload) + monkeypatch.setattr( + summary_all, + "build_quota_should_run", + lambda _status, *, goal_id, agent_id: {"ok": True, "goal_id": goal_id, "state": "eligible"}, + ) + + +def test_global_summary_from_a_project_registry_is_marked_incomplete(tmp_path: Path, monkeypatch) -> None: + _patch_status(monkeypatch, _global_status_payload(tmp_path, excluded=3)) + + payload = summary_all.build_summary_all( + registry_path=tmp_path / "registry.json", + runtime_root_override=None, + scan_roots=[], + agent_id=None, + time_range="24h", + limit=5, + ) + + envelope = payload["projection_envelope"] + assert envelope["projection"] == "global_summary" + assert envelope["coverage"]["scope"] == "global" + assert (envelope["coverage"]["expected_count"], envelope["coverage"]["included_count"]) == (4, 1) + assert envelope["coverage"]["omitted"][0]["reason"] == "outside_current_registry" + assert envelope["complete"] is False + assert "incomplete_coverage" in envelope["alert_reasons"] + assert envelope["upstream"][0]["projection"] == "status" + assert "status/registry" in _sources(envelope) + assert _sources(envelope)["goal_quota"]["item_count"] == 1 + markdown = summary_all.render_summary_all_markdown(payload) + assert "🔴 projection alerts: `incomplete_coverage`" in markdown + assert str(tmp_path) not in json.dumps(envelope) + + +def test_global_summary_from_the_global_registry_is_complete(tmp_path: Path, monkeypatch) -> None: + _patch_status(monkeypatch, _global_status_payload(tmp_path, excluded=0)) + + payload = summary_all.build_summary_all( + registry_path=tmp_path / "registry.json", + runtime_root_override=None, + scan_roots=[], + agent_id=None, + time_range="24h", + limit=5, + ) + + assert payload["projection_envelope"]["complete"] is True + assert payload["projection_envelope"]["alert"] is False + + +def test_global_gates_counts_unavailable_quota_and_missing_status_envelope(tmp_path: Path, monkeypatch) -> None: + status_payload = _global_status_payload(tmp_path, excluded=0) + status_payload.pop("projection_envelope") + monkeypatch.setattr(summary_all, "collect_status", lambda **_: status_payload) + monkeypatch.setattr( + summary_all, + "build_quota_should_run", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("quota ledger unreadable")), + ) + + payload = summary_all.build_global_gates( + registry_path=tmp_path / "registry.json", + runtime_root_override=None, + scan_roots=[], + agent_id=None, + limit=5, + ) + + envelope = payload["projection_envelope"] + sources = _sources(envelope) + assert sources["goal_quota"]["unreadable_count"] == 1 + assert sources["status"]["status"] == "not_read" + assert envelope["alert_reasons"] == ["missing_required_sources", "unreadable_sources"] + assert "🔴" in summary_all.render_global_gates_markdown(payload) From bea831d2c8ac7fb8f08eef5c5b6fd5fe1d8b5a04 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 26 Sep 2026 21:41:51 +0800 Subject: [PATCH 2/4] fix(projection): preserve unknown membership and qualify status output budget Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../contracts/interface-budget-contract.md | 23 +++++++- .../contracts/projection-envelope-contract.md | 5 ++ .../hot-path-interface-budget-smoke.py | 34 ++++------- .../goals/global_registry_health.py | 20 ++++++- loopx/control_plane/projection_envelope.ts | 15 +++-- loopx/control_plane/status/collection.py | 2 +- .../project_registry_io_manifest_v1.json | 38 +++++++++--- loopx/summary_all.py | 38 +++++++++--- .../projection_envelope.test.ts | 13 ++++ tests/test_projection_envelope.py | 59 ++++++++++++++++++- 10 files changed, 202 insertions(+), 45 deletions(-) diff --git a/docs/reference/contracts/interface-budget-contract.md b/docs/reference/contracts/interface-budget-contract.md index ee53ca079b..042b537bca 100644 --- a/docs/reference/contracts/interface-budget-contract.md +++ b/docs/reference/contracts/interface-budget-contract.md @@ -11,7 +11,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 <= 14500` | `nested_keys <= 360` | `top_level_keys <= 52` | -| `dashboard_status_json` | operator dashboard | render first-screen operator state | `history`, run artifacts, or project-local adapter output | `json_chars <= 19500` | `nested_keys <= 260` | `top_level_keys <= 25` | +| `dashboard_status_json` | operator dashboard | render first-screen operator state | `history`, run artifacts, or project-local adapter output | `json_chars <= 22500` | `nested_keys <= 350` | `top_level_keys <= 27` | These four budgets measure compact machine payloads. For `heartbeat_prompt_json`, the measured payload is the actual @@ -241,3 +241,24 @@ RRULE, unchanged-state clear flag, and short identity/profile signatures needed to detect reset transitions. Full identity/profile snapshots stay off the hot path; use status, history, active state, or a focused regression fixture when debugging why a reset token changed. + +### Status projection envelope budget decision + +The unchanged dashboard fixture measured 19,455 compact JSON characters, 244 +nested keys and 25 top-level keys before the projection envelope; the initial +envelope measured 21,518 / 332 / 26. The old 19,500 / 260 / 25 ceilings were +regression budgets, not transport limits. The operator needs source read times, +read failures and scope coverage to distinguish a cached or partial observation +from a current, complete view. Per-source rows support diagnosis and replay; +removing them would lose that contract. The bounded five-source envelope is +retained rather than shortening names or shrinking the fixture. Ceilings become +22,500 / 350 / 27, leaving 982 characters, 18 nested keys and one top-level key +above the measured head for variation. Other hot surfaces retain their budgets. +This adds a read contract to default status; it grants no execution authority. + +同一 dashboard 负载在新增 envelope 前为 19,455 字符／244 个嵌套键／25 个顶层键， +初始 head 为 21,518／332／26。旧上限属于回归预算而非传输硬限制。操作员需要 +来源读取时间、错误与范围覆盖来识别缓存和部分观察；逐来源数据还支撑诊断和重放， +不能为过线删除。保留五个有界来源，不缩小负载或改短字段名，将上限同步调整为 +22,500／350／27，较实测 head 保留 982 字符、18 个嵌套键和一个顶层键的余量。 +其他热表面预算保持原值。默认 status 新增读合同，不授予执行权限。 diff --git a/docs/reference/contracts/projection-envelope-contract.md b/docs/reference/contracts/projection-envelope-contract.md index 1167459d4e..ce6fd7fa80 100644 --- a/docs/reference/contracts/projection-envelope-contract.md +++ b/docs/reference/contracts/projection-envelope-contract.md @@ -123,3 +123,8 @@ kernel on every hit, which restamps `served_at` and recomputes staleness. A cache record without an envelope, or with one the kernel decoder rejects, is a cache miss (`missing_projection_envelope` / `invalid_projection_envelope`), never an unlabeled hit. + +An unknown `expected_count` is never complete. Global views require global +membership even when project-local status treats that source as optional; +missing or unreadable membership alerts and preserves a null denominator. +After the source is repaired, a fresh collection can certify coverage again. diff --git a/examples/control_plane/hot-path-interface-budget-smoke.py b/examples/control_plane/hot-path-interface-budget-smoke.py index bb4b69953c..716a3c6acd 100644 --- a/examples/control_plane/hot-path-interface-budget-smoke.py +++ b/examples/control_plane/hot-path-interface-budget-smoke.py @@ -80,9 +80,9 @@ "owner": "operator dashboard", "consumer": "render first-screen operator state", "cold_path": "history, run artifacts, or project-local adapter output", - "max_json_chars": 19_500, - "max_nested_keys": 260, - "max_top_level_keys": 25, + "max_json_chars": 22_500, + "max_nested_keys": 350, + "max_top_level_keys": 27, }, } @@ -360,8 +360,8 @@ def assert_cadence_projection( assert projected["overdue"] is False, projected assert projected["within_budget"] is True, projected assert projected["next_check_due_at"] == "2099-01-02T00:10:00+00:00", projected - assert projected["headroom_remaining"] == 0, projected - assert projected["recommendation"] == "rerun_hot_path_interface_budget_smoke", projected + assert projected["headroom_remaining"] == 7, projected + assert projected["recommendation"] == "quiet_skip_until_next_check_due", projected quota_payload = build_quota_should_run( status_payload, @@ -450,26 +450,18 @@ def main() -> int: assert cadence["surface_count"] == len(summaries), cadence assert cadence["next_check_due_at"] == "2099-01-02T00:10:00+00:00", cadence assert cadence["minimum_headroom_ratio"] is not None, cadence - assert cadence["headroom_remaining"] == 0, cadence - assert cadence["recommendation"] == "rerun_hot_path_interface_budget_smoke", cadence - relaxed_summaries = [dict(summary) for summary in summaries] - for summary in relaxed_summaries: - if summary["json_chars"] == summary["max_json_chars"]: - summary["max_json_chars"] = summary["json_chars"] + 1 - if summary["nested_keys"] == summary["max_nested_keys"]: - summary["max_nested_keys"] = summary["nested_keys"] + 1 - if summary["top_level_keys"] == summary["max_top_level_keys"]: - summary["max_top_level_keys"] = summary["top_level_keys"] + 1 - relaxed_cadence = build_interface_budget_cadence( - relaxed_summaries, + assert cadence["headroom_remaining"] == 7, cadence + assert cadence["recommendation"] == "quiet_skip_until_next_check_due", cadence + saturated_summaries = [dict(summary) for summary in summaries] + saturated_summaries[0]["max_json_chars"] = saturated_summaries[0]["json_chars"] + saturated_cadence = build_interface_budget_cadence( + saturated_summaries, checked_at="2099-01-01T00:10:00+00:00", now="2099-01-01T01:00:00+00:00", freshness_hours=24, ) - assert relaxed_cadence["within_budget"] is True, relaxed_cadence - assert relaxed_cadence["overdue"] is False, relaxed_cadence - assert relaxed_cadence["headroom_remaining"] > 0, relaxed_cadence - assert relaxed_cadence["recommendation"] == "quiet_skip_until_next_check_due", relaxed_cadence + assert saturated_cadence["headroom_remaining"] == 0, saturated_cadence + assert saturated_cadence["recommendation"] == "rerun_hot_path_interface_budget_smoke", saturated_cadence stale_cadence = build_interface_budget_cadence( summaries, checked_at="2099-01-01T00:10:00+00:00", diff --git a/loopx/control_plane/goals/global_registry_health.py b/loopx/control_plane/goals/global_registry_health.py index ce01830549..b72adee40f 100644 --- a/loopx/control_plane/goals/global_registry_health.py +++ b/loopx/control_plane/goals/global_registry_health.py @@ -47,6 +47,7 @@ def collect_global_registry_health( if not global_path.exists(): return { "available": False, + "read_status": "missing", "ok": True, "registry": str(global_path), "current_registry": str(registry_path), @@ -56,7 +57,24 @@ def collect_global_registry_health( "checks": [], } - global_registry = load_registry(global_path) + try: + global_registry = load_registry(global_path) + except (OSError, ValueError): + return { + "available": False, + "read_status": "unreadable", + "ok": False, + "registry": str(global_path), + "current_registry": str(registry_path), + "current_registry_is_global": False, + "summary": {"high": 1, "action": 0, "info": 0, "checks": 1, "findings": 1}, + "findings": [global_registry_finding( + kind="global_registry_unreadable", severity="high", + message="global membership could not be read", + recommended_action="repair the global registry source and retry the projection", + )], + "checks": ["global registry readability"], + } global_goals = registry_goals(global_registry) current_goals = registry_goals(current_registry) current_ids = {str(goal.get("id")) for goal in current_goals if goal.get("id")} diff --git a/loopx/control_plane/projection_envelope.ts b/loopx/control_plane/projection_envelope.ts index aae6a1988f..d1637d7beb 100644 --- a/loopx/control_plane/projection_envelope.ts +++ b/loopx/control_plane/projection_envelope.ts @@ -200,6 +200,13 @@ function sealSource(fact: SourceFact, servedMillis: number): JsonObject { }; } +function coverageComplete(coverage: CoverageFact, upstream: UpstreamSummary[]): boolean { + // An unknown denominator can never certify the requested scope. + return coverage.expected_count !== null && coverage.omitted.length === 0 + && coverage.included_count >= coverage.expected_count + && upstream.every((row) => row.complete); +} + function sealFacts( projection: string, observed: { text: string; millis: number }, @@ -218,9 +225,7 @@ function sealFacts( const sources = facts.map((fact) => sealSource(fact, served.millis)); const truncated = coverage.shown_count !== null && coverage.available_count !== null && coverage.available_count > coverage.shown_count; - const complete = coverage.omitted.length === 0 - && (coverage.expected_count === null || coverage.included_count >= coverage.expected_count) - && upstream.every((row) => row.complete); + const complete = coverageComplete(coverage, upstream); const reasons = new Set(); const alertSourceIds: string[] = []; for (const row of sources) { @@ -281,9 +286,7 @@ function decodeSealedEnvelope(value: unknown, label: string): { .map(decodeUpstreamSummary); const projection = identifier(envelope.projection, `${label}.projection`); const observed = timestamp(envelope.observed_at, `${label}.observed_at`); - const complete = coverage.omitted.length === 0 - && (coverage.expected_count === null || coverage.included_count >= coverage.expected_count) - && upstream.every((row) => row.complete); + const complete = coverageComplete(coverage, upstream); return { projection, observed, sources, coverage, upstream, complete }; } diff --git a/loopx/control_plane/status/collection.py b/loopx/control_plane/status/collection.py index 35d695eddb..ec216fe5f2 100644 --- a/loopx/control_plane/status/collection.py +++ b/loopx/control_plane/status/collection.py @@ -277,7 +277,7 @@ def collect_status( source_fact("registry", last_read_at=registry_read_at, item_count=len(registry_goals(registry))), source_fact( "global_registry", - read_status="read" if global_registry.get("available") else "missing", + read_status="read" if global_registry.get("available") else global_registry.get("read_status", "missing"), last_read_at=global_registry_read_at if global_registry.get("available") else None, required=False, item_count=global_registry.get("global_goal_count"), diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index c8b4f7eb37..45357716b3 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -437,10 +437,34 @@ "api": "load_registry", "classification": "codec_api" }, + { + "site": "loopx/cli_commands/authority_archive.py::.authority_upgrade_roots::codec_read:load_registry#1", + "line": 86, + "column": 16, + "kind": "codec_read", + "api": "load_registry", + "classification": "codec_api" + }, + { + "site": "loopx/cli_commands/authority_archive.py::.authority_upgrade_roots::codec_read:load_registry#2", + "line": 94, + "column": 27, + "kind": "codec_read", + "api": "load_registry", + "classification": "codec_api" + }, + { + "site": "loopx/cli_commands/authority_archive.py::.authority_upgrade_roots::codec_read:load_registry#3", + "line": 104, + "column": 48, + "kind": "codec_read", + "api": "load_registry", + "classification": "codec_api" + }, { "site": "loopx/cli_commands/authority_archive.py::.handle_authority_archive_command::codec_read:load_registry#1", - "line": 49, - "column": 13, + "line": 64, + "column": 17, "kind": "codec_read", "api": "load_registry", "classification": "codec_api" @@ -799,7 +823,7 @@ }, { "site": "loopx/cli_commands/todo_continuation.py::.handle_todo_continuation::codec_read:load_registry#1", - "line": 136, + "line": 140, "column": 20, "kind": "codec_read", "api": "load_registry", @@ -1151,8 +1175,8 @@ }, { "site": "loopx/control_plane/goals/global_registry_health.py::.collect_global_registry_health::codec_read:load_registry#1", - "line": 59, - "column": 23, + "line": 61, + "column": 27, "kind": "codec_read", "api": "load_registry", "classification": "codec_api" @@ -1463,7 +1487,7 @@ }, { "site": "loopx/control_plane/work_items/task_lease_acquire_adapter.py::.execute_native_task_lease_acquire::codec_read:load_registry#1", - "line": 397, + "line": 400, "column": 20, "kind": "codec_read", "api": "load_registry", @@ -1471,7 +1495,7 @@ }, { "site": "loopx/control_plane/work_items/task_lease_acquire_adapter.py::.execute_native_task_lease_lifecycle::codec_read:load_registry#1", - "line": 709, + "line": 715, "column": 24, "kind": "codec_read", "api": "load_registry", diff --git a/loopx/summary_all.py b/loopx/summary_all.py index e1e16ae754..a9529f0c1e 100644 --- a/loopx/summary_all.py +++ b/loopx/summary_all.py @@ -141,7 +141,7 @@ def _global_projection_envelope( projection: str, status_payload: dict[str, Any], *, - quota_read_at: str, + quota_read_at: str | None, quota_evaluated: int, quota_unavailable: int, shown_count: int, @@ -158,13 +158,29 @@ def _global_projection_envelope( excluded = int(global_registry.get("current_registry_excluded_goal_count") or 0) refs = [str(ref) for ref in _as_list(global_registry.get("current_registry_excluded_goal_ids"))] else: - expected = int(status_coverage.get("expected_count") or 0) + expected = None # Local membership cannot prove global membership. + global_source = next( + (row for row in _as_list(status_envelope.get("sources")) + if isinstance(row, dict) and row.get("source_id") == "global_registry"), + {}, + ) return seal_projection_envelope( projection=projection, observed_at=_now_iso(), sources=[ + source_fact( + "global_registry", + read_status=str(global_source.get("read_status") or ( + "read" if global_registry.get("available") else "not_read" + )), + last_read_at=global_source.get("last_read_at") or ( + quota_read_at if global_registry.get("available") else None + ), + item_count=expected, + ), source_fact( "goal_quota", + read_status="read" if quota_read_at else "not_read", last_read_at=quota_read_at, item_count=quota_evaluated, unreadable_count=quota_unavailable, @@ -174,7 +190,8 @@ def _global_projection_envelope( coverage={ "scope": "global", "expected_count": expected, - "included_count": max(0, expected - excluded), + "included_count": (max(0, expected - excluded) if expected is not None + else int(status_coverage.get("included_count") or 0)), "omitted": ( [{"reason": "outside_current_registry", "count": excluded, "refs": refs[:8]}] if excluded @@ -455,7 +472,12 @@ def build_global_gates( limit=normalized_limit, ) if status_payload.get("ok") is not True: - return build_global_gates_error("Global status source unavailable.") + payload = build_global_gates_error("Global status source unavailable.") + payload["projection_envelope"] = _global_projection_envelope( + "global_gates", status_payload, quota_read_at=None, + quota_evaluated=0, quota_unavailable=0, shown_count=0, available_count=0, + ) + return payload quota_read_at = _now_iso() state = _collect_global_gate_state( status_payload, @@ -545,9 +567,11 @@ def build_global_gates_error(error: object) -> dict[str, Any]: def render_global_gates_markdown(payload: dict[str, Any]) -> str: if not payload.get("ok"): - return "# LoopX Global Gates\n\n- ok: `False`\n- error: " + _redact_text( - payload.get("error") - ) + lines = ["# LoopX Global Gates", "", "- ok: `False`", + "- error: " + _redact_text(payload.get("error"))] + if payload.get("projection_envelope"): + lines.extend(render_projection_envelope_markdown(payload["projection_envelope"])) + return "\n".join(lines) summary = _as_dict(payload.get("summary")) lines = [ diff --git a/tests/control_plane_ts/projection_envelope.test.ts b/tests/control_plane_ts/projection_envelope.test.ts index 871faa27c6..ea70fc032a 100644 --- a/tests/control_plane_ts/projection_envelope.test.ts +++ b/tests/control_plane_ts/projection_envelope.test.ts @@ -180,3 +180,16 @@ test("the decoder rejects values that cannot establish freshness", () => { schema_version: PROJECTION_ENVELOPE_SERVE_REQUEST, envelope: { schema_version: "other" }, served_at: OBSERVED, }), /schema mismatch/); }); + +test("unknown coverage stays incomplete through replay and upstream composition", () => { + const unknown = sealProjectionEnvelope(sealRequest({ + coverage: { scope: "global", expected_count: null, included_count: 1 }, + })); + assert.equal(unknown.complete, false); + assert.deepEqual(unknown.alert_reasons, ["incomplete_coverage"]); + const replay = sealProjectionEnvelope({ schema_version: PROJECTION_ENVELOPE_SERVE_REQUEST, + envelope: unknown, served_at: at(1) }); + assert.equal(replay.complete, false); + const derived = sealProjectionEnvelope(sealRequest({ projection: "global_gates", upstream: [unknown] })); + assert.equal(derived.complete, false); +}); diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py index bc215c6ad3..5f3754f6a0 100644 --- a/tests/test_projection_envelope.py +++ b/tests/test_projection_envelope.py @@ -29,6 +29,7 @@ def _registry(tmp_path: Path, goal_ids: tuple[str, ...] = ("goal-a", "goal-b")) { "id": goal_id, "objective": "Project envelope fixture.", + "domain": "software-development", "repo": str(tmp_path), "state_file": str(tmp_path / "ACTIVE_GOAL_STATE.md"), "adapter": {"kind": "read_only_project_map_v0"}, @@ -165,7 +166,15 @@ def test_cache_entry_with_a_tampered_envelope_is_a_miss(tmp_path: Path) -> None: def _global_status_payload(tmp_path: Path, *, excluded: int) -> dict: - payload = _status(_registry(tmp_path, ("goal-a",))) + registry = _registry(tmp_path, ("goal-a",)) + runtime = tmp_path / "runtime" + runtime.mkdir(exist_ok=True) + global_members = json.loads(registry.read_text()) + global_members["goals"].extend( + {**global_members["goals"][0], "id": f"other-{index}"} for index in range(excluded) + ) + (runtime / "registry.global.json").write_text(json.dumps(global_members)) + payload = _status(registry) payload["ok"] = True payload["global_registry"] = { "available": True, @@ -257,3 +266,51 @@ def test_global_gates_counts_unavailable_quota_and_missing_status_envelope(tmp_p assert sources["status"]["status"] == "not_read" assert envelope["alert_reasons"] == ["missing_required_sources", "unreadable_sources"] assert "🔴" in summary_all.render_global_gates_markdown(payload) + + +def test_real_cli_global_membership_failure_and_recovery(tmp_path): + """Global scope must never use local membership as its denominator.""" + import subprocess + import sys + + registry = _registry(tmp_path, ("goal-a",)) + runtime = tmp_path / "runtime" + runtime.mkdir(exist_ok=True) + global_path = runtime / "registry.global.json" + global_registry = json.loads(registry.read_text()) + second_state = tmp_path / "SECOND_GOAL_STATE.md" + second_state.write_text("# Second Goal\n") + global_registry["goals"].append({**global_registry["goals"][0], "id": "goal-b", + "state_file": str(second_state)}) + + def cli(command, registry_path=registry): + result = subprocess.run( + [sys.executable, "-c", "from loopx.cli_runtime import main; raise SystemExit(main())", + "--registry", str(registry_path), "--format", "json", command, "--limit", "1"], + text=True, capture_output=True, timeout=30, + ) + assert result.returncode in (0, 1), result.stderr + return json.loads(result.stdout) + + + for content, reason in ((None, "missing_required_sources"), ("{broken", "unreadable_sources")): + if content is None: + global_path.unlink(missing_ok=True) + else: + global_path.write_text(content) + for command in ("global-summary", "global-gates"): + envelope = cli(command)["projection_envelope"] + assert envelope["coverage"]["expected_count"] is None + assert envelope["complete"] is False and envelope["alert"] is True + assert reason in envelope["alert_reasons"] + assert "incomplete_coverage" in envelope["alert_reasons"] + if content is None: + local = cli("status")["projection_envelope"] + assert local["complete"] is True and local["alert"] is False + global_path.write_text(json.dumps(global_registry)) + for command in ("global-summary", "global-gates"): + local = cli(command)["projection_envelope"] + assert (local["coverage"]["included_count"], local["coverage"]["expected_count"]) == (1, 2) + assert local["complete"] is False + whole = cli(command, global_path)["projection_envelope"] + assert whole["complete"] is True and whole["alert"] is False From 56ae504ba36a9dca442b5c38d6402509c31a729e Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 26 Sep 2026 21:49:46 +0800 Subject: [PATCH 3/4] test(projection): qualify one-time emitted envelope growth at status boundary Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../contracts/interface-budget-contract.md | 16 +++++++++ .../control_plane/cli-output-probe-runner.py | 3 ++ .../projection_envelope_facts.py | 2 +- .../testing/cli_output_differential.py | 24 ++++++++++++++ .../testing/cli_output_semantics.py | 6 ++++ .../test_cli_output_differential.py | 33 +++++++++++++++++++ 6 files changed, 83 insertions(+), 1 deletion(-) diff --git a/docs/reference/contracts/interface-budget-contract.md b/docs/reference/contracts/interface-budget-contract.md index 042b537bca..deec4a3629 100644 --- a/docs/reference/contracts/interface-budget-contract.md +++ b/docs/reference/contracts/interface-budget-contract.md @@ -262,3 +262,19 @@ This adds a read contract to default status; it grants no execution authority. 不能为过线删除。保留五个有界来源，不缩小负载或改短字段名，将上限同步调整为 22,500／350／27，较实测 head 保留 982 字符、18 个嵌套键和一个顶层键的余量。 其他热表面预算保持原值。默认 status 新增读合同，不授予执行权限。 + +The emitted CLI matrix separately measured +2,801 pretty JSON characters, ++102 lines and +1,910 compact characters on small, crowded and multi-agent +status fixtures. Markdown added 127 characters before the explicit schema +marker. The existing schema-transition mechanism grants **only status and its +explicit task-graph variant**, and only `none -> loopx_projection_envelope_v0`, +3,000 JSON chars/bytes, 110 lines and 2,048 compact chars; Markdown receives +192 chars/224 bytes and three lines. The marker makes this transition visible +and review-required. Unknown schemas, reverse transitions, unrelated surfaces +and subsequent v0 growth retain ordinary budgets. Absolute ceilings stay intact. + +CLI 同负载差分另测得 JSON 增加 2,801 字符、102 行、1,910 个紧凑字符；Markdown +在显式 schema 标识前增加 127 字符。沿用既有 schema 迁移预算机制，仅 status 及 +其 task-graph 显式变体的 none → v0 获得一次 3,000 JSON 字符／字节、110 行、 +2,048 紧凑字符余量；Markdown 余量为 192 字符／224 字节和三行。该迁移必须评审。 +未知 schema、反向迁移、其他表面和后续 v0 增长使用普通预算，绝对上限保持不变。 diff --git a/examples/control_plane/cli-output-probe-runner.py b/examples/control_plane/cli-output-probe-runner.py index e8b89b1159..e9d3a0b061 100644 --- a/examples/control_plane/cli-output-probe-runner.py +++ b/examples/control_plane/cli-output-probe-runner.py @@ -119,6 +119,9 @@ def _receipt_row( if isinstance(payload, dict) else [] ), + "projection_envelope_schema_versions": ( + semantics.projection_envelope_schema_versions(payload if isinstance(payload, dict) else text) + ), "todo_work_counts_schema_versions": ( semantics.todo_work_counts_schema_versions(payload) if isinstance(payload, dict) diff --git a/loopx/control_plane/projection_envelope_facts.py b/loopx/control_plane/projection_envelope_facts.py index 8c908a0a5e..01665f43ad 100644 --- a/loopx/control_plane/projection_envelope_facts.py +++ b/loopx/control_plane/projection_envelope_facts.py @@ -89,7 +89,7 @@ def render_projection_envelope_markdown(envelope: Any) -> list[str]: fresh_count = sum(1 for row in sources if row.get("status") == "fresh") marker = f"{ALERT_MARKER} " if envelope.get("alert") else "" line = ( - f"- projection: {marker}observed_at=`{envelope.get('observed_at')}` " + f"- projection: {marker}envelope=`{PROJECTION_ENVELOPE_SCHEMA_VERSION}` observed_at=`{envelope.get('observed_at')}` " f"age=`{envelope.get('age_seconds')}s`" + (" (cached)" if envelope.get("served_from_cache") else "") + f" sources_fresh=`{fresh_count}/{len(sources) - absent_optional}`" diff --git a/loopx/control_plane/testing/cli_output_differential.py b/loopx/control_plane/testing/cli_output_differential.py index a8af349b04..ed690947ac 100644 --- a/loopx/control_plane/testing/cli_output_differential.py +++ b/loopx/control_plane/testing/cli_output_differential.py @@ -622,6 +622,16 @@ def _schema_migration_growth_allowance( return max(allowances, default=0) +# 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. +_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}, +) + + def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]: row_id = str(base["row_id"]) failures: list[str] = [] @@ -656,6 +666,19 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A _runtime_root_route_growth_allowances(base, candidate) ) + base_projection = tuple(base.get("projection_envelope_schema_versions") or []) + candidate_projection = tuple(candidate.get("projection_envelope_schema_versions") or []) + projection_migration = ( + (row_id.startswith("surface/status/") or row_id.startswith("variant/status_task_graph_detail/")) + and base_projection == () and candidate_projection == ("loopx_projection_envelope_v0",) + ) + if base_projection != candidate_projection: + if projection_migration: + review_signals.append("projection envelope schema migrated: none -> loopx_projection_envelope_v0") + else: + failures.append("projection envelope schema coverage changed") + projection_allowance = (_PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE.json + if output_format == "json" else _PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE.markdown) deltas: dict[str, int | None] = {} allowances: dict[str, int | None] = {} for metric in ("chars", "utf8_bytes", "lines", "compact_payload_chars"): @@ -692,6 +715,7 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A metric, ), _schema_migration_growth_allowance(migration, metric), + projection_allowance[metric] if projection_migration else 0, ) # Thin installed prompts contain bilingual lifecycle instructions. A # small character-level clarification can cost three bytes per CJK diff --git a/loopx/control_plane/testing/cli_output_semantics.py b/loopx/control_plane/testing/cli_output_semantics.py index 542fcb62ba..7b04a8e2c1 100644 --- a/loopx/control_plane/testing/cli_output_semantics.py +++ b/loopx/control_plane/testing/cli_output_semantics.py @@ -204,3 +204,9 @@ def markdown_headings(text: str) -> list[str]: def runtime_root_command_route_count(text: str) -> int: return len(_RUNTIME_ROOT_COMMAND_ROUTE.findall(text)) + + +def projection_envelope_schema_versions(value: Any) -> list[str]: + if isinstance(value, str): + return sorted(set(re.findall(r"^- projection: (?:🔴 )?envelope=`([a-z0-9_]+)`", value, re.MULTILINE))) + return _schema_versions_for_key(value, "projection_envelope") diff --git a/tests/control_plane/test_cli_output_differential.py b/tests/control_plane/test_cli_output_differential.py index ae8c50af43..eb65804fe7 100644 --- a/tests/control_plane/test_cli_output_differential.py +++ b/tests/control_plane/test_cli_output_differential.py @@ -1076,3 +1076,36 @@ def test_measurement_only_probe_skips_ceiling_but_keeps_semantic_shape() -> None semantic_json_keys=("required",), markdown_anchor=None, ) + + +@pytest.mark.parametrize("output_format", ["json", "markdown"]) +def test_projection_envelope_migration_is_status_only_bounded_and_one_time(output_format): + 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} + 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}") + 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)) + assert result["ok"] and result["review_required"] + # JSON fixtures are large enough for ordinary ratio allowances; use their + # smaller real scale when checking bounded Markdown growth. + if output_format == "markdown": + base.update(chars=1000, utf8_bytes=1000, lines=30) + candidate.update(**{metric: base[metric] + limit for metric, limit in limits.items()}) + for metric in ("chars", "utf8_bytes", "lines"): + too_large = {**candidate, metric: candidate[metric] + 1} + assert not compare_cli_output_receipts(_receipt(base), _receipt(too_large))["ok"] + grown = {**candidate, "chars": candidate["chars"] + limits["chars"]} + assert not compare_cli_output_receipts(_receipt(candidate), _receipt(grown))["ok"] + 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"]} + assert not compare_cli_output_receipts(_receipt(outside_base), _receipt(outside))["ok"] From 9e839bd9a84be0f5b6f0c18fd3bdcc166dde642b Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 26 Sep 2026 22:07:08 +0800 Subject: [PATCH 4/4] refactor(projection): isolate output migration qualification rule Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../testing/cli_output_differential.py | 35 ++++++++++++------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/loopx/control_plane/testing/cli_output_differential.py b/loopx/control_plane/testing/cli_output_differential.py index ed690947ac..b883abcb80 100644 --- a/loopx/control_plane/testing/cli_output_differential.py +++ b/loopx/control_plane/testing/cli_output_differential.py @@ -632,6 +632,23 @@ def _schema_migration_growth_allowance( ) +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.""" + 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/")) + 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) + return allowance, [], ["projection envelope schema migrated: none -> loopx_projection_envelope_v0"] + return {}, ["projection envelope schema coverage changed"], [] + + def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]: row_id = str(base["row_id"]) failures: list[str] = [] @@ -666,19 +683,11 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A _runtime_root_route_growth_allowances(base, candidate) ) - base_projection = tuple(base.get("projection_envelope_schema_versions") or []) - candidate_projection = tuple(candidate.get("projection_envelope_schema_versions") or []) - projection_migration = ( - (row_id.startswith("surface/status/") or row_id.startswith("variant/status_task_graph_detail/")) - and base_projection == () and candidate_projection == ("loopx_projection_envelope_v0",) + projection_allowance, projection_failures, projection_signals = _projection_envelope_migration( + base, candidate, output_format=output_format, ) - if base_projection != candidate_projection: - if projection_migration: - review_signals.append("projection envelope schema migrated: none -> loopx_projection_envelope_v0") - else: - failures.append("projection envelope schema coverage changed") - projection_allowance = (_PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE.json - if output_format == "json" else _PROJECTION_ENVELOPE_V0_MIGRATION_ALLOWANCE.markdown) + failures.extend(projection_failures) + review_signals.extend(projection_signals) deltas: dict[str, int | None] = {} allowances: dict[str, int | None] = {} for metric in ("chars", "utf8_bytes", "lines", "compact_payload_chars"): @@ -715,7 +724,7 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A metric, ), _schema_migration_growth_allowance(migration, metric), - projection_allowance[metric] if projection_migration else 0, + projection_allowance.get(metric, 0), ) # Thin installed prompts contain bilingual lifecycle instructions. A # small character-level clarification can cost three bytes per CJK