diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md index f93211e081..3a10b0bd83 100644 --- a/docs/reference/protocols/turn-envelope-v0.md +++ b/docs/reference/protocols/turn-envelope-v0.md @@ -538,6 +538,42 @@ work context 中已加载的来源，其次保留 planning horizon 或第一份 签名覆盖的 work context，拒绝缺失或不可用来源。这只证明展示成本下降，尚不证明 上游读取减少、模型采用或完整任务效率提升。 +Known Todo diagnostic lanes share `quota/diagnostic_projection.ts` through the +existing coarse context call. Read-only status/plan projects all rows in one +call, rather than a call per Goal or lane. Unknown lists, fields, versions and +complex nested item content remain full; field recognition reuses the canonical +Todo contract. Retained items, capability candidates, portfolio text, planning +horizon and selected-work sources are not shortened. Explicit detail returns the +original summaries. Upstream compaction remains visible as `source_projection`. +Status/plan adds one projection request; Agent/vision detail still needs one +when the User summary remains compact. Explicit full Todo roles bypass that +diagnostic request. These transport costs must be included in comparisons. +Diagnostic replies carry only display edits and references to the immutable +input. The Python adapter applies them to its original facts, preserving float +spelling, negative zero, large-integer precision and relocated upstream metadata. +Reference-only callers keep their existing transport; no Python compaction +decision is reintroduced. The read codec's `content_revision` is recognized as +known display metadata and remains on retained items. +The legacy detail command observes current state and cannot recover an earlier +decision. The scoped native consumer and original-capture reference are a separate +[controller integration proposal](https://github.com/loopx-project/loopx/pull/7430), +which reuses this same owner. This does not change admission, claims, leases, +settlement or native context defaults; it does not qualify net efficiency gains. + +中文：已知 Todo 诊断列表通过既有粗粒度上下文调用复用同一 typed owner；只读 +status/plan 整批行只调用一次，不逐 Goal 或列表调用。未知列表、字段、版本和 +复杂嵌套条目保留全文，字段识别复用规范 Todo 契约；保留条目、候选、portfolio +正文、规划视野和当前工作来源不截短，详情返回原摘要，上游压缩信息保留于 +`source_projection`。旧详情命令读取当前状态，不能恢复历史决定。 +status/plan 新增一次投影请求；Agent/vision 详情若仍压缩 User 摘要，也需一次； +显式展开两类 Todo 则跳过诊断投影。这些传输成本须计入比较。scoped 原生 +消费者及同 capture 引用属于单独的 controller 集成提案 #7430，复用同一 owner。 +此处不改准入、claim、lease、结算或原生默认，也不代表完整效率收益通过。 +诊断回复只传展示变更及不可变输入的引用；Python adapter 在原始事实上应用， +保留浮点表示、负零、大整数精度及迁移后的上游元数据。仅引用的旧调用方保持 +原传输行为，不重新引入 Python 裁剪决定。读取 codec 已有的 `content_revision` +属于已知展示元数据，在保留条目上仍完整存在。 + The selected-Todo display body is display-only and may be bounded; the typed owner validates the exact read against selected identity, lifecycle, claim and source revision. A supplied full-body snapshot must match before fulfilling the read. If the full body is already diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py index 58c8868719..45ef36f048 100644 --- a/loopx/control_plane/quota/cli_projection.py +++ b/loopx/control_plane/quota/cli_projection.py @@ -1,20 +1,15 @@ from __future__ import annotations import json +from copy import deepcopy from typing import Any from .decision_summary import compact_quota_decision from .should_run_packet import reference_duplicate_fallback_guidance -QUOTA_CLI_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION = ( - "quota_cli_todo_summary_compaction_v0" -) QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND = ( "quota should-run --include-detail agent-todos" ) -QUOTA_CLI_USER_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION = ( - "quota_cli_user_todo_summary_compaction_v0" -) QUOTA_CLI_USER_TODO_SUMMARY_DETAIL_COMMAND = ( "quota should-run --include-detail user-todos" ) @@ -31,66 +26,12 @@ QUOTA_CLI_REPLAN_ACTION_COMPACTION_SCHEMA_VERSION = ( "quota_cli_replan_action_compaction_v0" ) -QUOTA_CLI_CAPABILITY_GATE_COMPACTION_SCHEMA_VERSION = ( - "quota_cli_capability_gate_compaction_v0" -) -QUOTA_CLI_ACTION_PORTFOLIO_COMPACTION_SCHEMA_VERSION = ( - "quota_cli_action_portfolio_compaction_v1" -) QUOTA_CLI_MONITOR_POLL_DECISION_COMPACTION_SCHEMA_VERSION = ( "quota_cli_monitor_poll_decision_compaction_v0" ) QUOTA_CLI_MONITOR_POLL_DECISION_DETAIL_COMMAND = ( "quota monitor-poll --include-detail decisions" ) -_RETAINED_AGENT_ITEM_LANES = { - "first_executable_items": 3, - "unclaimed_priority_open_items": 3, - "monitor_due_items": 1, - "monitor_capability_blocked_due_items": 2, - "monitor_schedule_gap_items": 1, - "current_agent_blocker_items": 2, -} -_RETAINED_USER_ITEM_LANES = { - "first_open_items": 3, - "gate_open_items": 3, - "active_next_action_items": 3, -} -_RETAINED_AGENT_ITEM_FIELDS = ( - "schema_version", - "todo_id", - "index", - "text", - "status", - "priority", - "task_class", - "action_kind", - "task_domain", - "claimed_by", - "bound_agent", - "goal_bound", - "blocks_agent", - "global_gate", - "task_repository", - "required_capabilities", - "required_write_scopes", - "excluded_agents", - "unblocks_todo_id", - "continuation_policy", - "resume_when", - "resume_monitor_generation", - "target_key", - "cadence", - "next_due_at", - "expires_at", - "last_checked_at", - "consecutive_no_change", - "material_change", - "material_change_generation", - "result_hash", - "reason", -) -_RETAINED_SUCCESSION_WARNING_TODO_IDS = 3 _RUNTIME_CAPABILITY_PREFIX_FIELDS = ( "ok", "status_health_ok", @@ -261,115 +202,6 @@ def compact_quota_monitor_poll_cli_payload( return projected -def _compact_agent_item(item: Any) -> Any: - if not isinstance(item, dict): - return item - return { - key: item[key] - for key in _RETAINED_AGENT_ITEM_FIELDS - if key in item - } - - -def _compact_nested_item_lists( - value: dict[str, Any], - *, - omitted_lanes: dict[str, int], - path: str, -) -> dict[str, Any]: - compact: dict[str, Any] = {} - for key, child in value.items(): - if isinstance(child, list) and key.endswith("items"): - if path == "todo_succession_warning": - todo_ids = [ - str(item.get("todo_id")) - for item in child - if isinstance(item, dict) and item.get("todo_id") - ][:_RETAINED_SUCCESSION_WARNING_TODO_IDS] - if todo_ids: - compact["todo_ids"] = todo_ids - if child: - omitted_lanes[f"{path}.{key}"] = len(child) - continue - compact[key] = child - return compact - - -def _compact_agent_todo_summary( - summary: dict[str, Any], *, detail_command: str = QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND -) -> dict[str, Any]: - compact: dict[str, Any] = {} - omitted_lanes: dict[str, int] = {} - for key, value in summary.items(): - if isinstance(value, list): - limit = _RETAINED_AGENT_ITEM_LANES.get(key) - if limit is None: - if value: - omitted_lanes[key] = len(value) - continue - compact[key] = [ - _compact_agent_item(item) - for item in value[:limit] - ] - if len(value) > limit: - omitted_lanes[key] = len(value) - limit - continue - if isinstance(value, dict): - compact[key] = _compact_nested_item_lists( - value, - omitted_lanes=omitted_lanes, - path=key, - ) - continue - compact[key] = value - - compact["payload_compaction"] = { - "schema_version": QUOTA_CLI_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION, - "retained_item_lanes": sorted( - lane - for lane in _RETAINED_AGENT_ITEM_LANES - if lane != "current_agent_blocker_items" or summary.get(lane) - ), - "omitted_lanes": omitted_lanes, - "full_detail_cold_path": detail_command, - } - return compact - - -def _compact_user_todo_summary( - summary: dict[str, Any], *, detail_command: str = QUOTA_CLI_USER_TODO_SUMMARY_DETAIL_COMMAND -) -> dict[str, Any]: - compact: dict[str, Any] = {} - omitted_lanes: dict[str, int] = {} - for key, value in summary.items(): - if isinstance(value, list): - limit = _RETAINED_USER_ITEM_LANES.get(key) - if limit is None: - if value: - omitted_lanes[key] = len(value) - continue - compact[key] = value[:limit] - if len(value) > limit: - omitted_lanes[key] = len(value) - limit - continue - if isinstance(value, dict): - compact[key] = _compact_nested_item_lists( - value, - omitted_lanes=omitted_lanes, - path=key, - ) - continue - compact[key] = value - - compact["payload_compaction"] = { - "schema_version": QUOTA_CLI_USER_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION, - "retained_item_lanes": sorted(_RETAINED_USER_ITEM_LANES), - "omitted_lanes": omitted_lanes, - "full_detail_cold_path": detail_command, - } - return compact - - def _compact_goal_boundary(boundary: dict[str, Any]) -> dict[str, Any]: authority = boundary.get("checkpointed_boundary_authority") if not isinstance(authority, dict) or not isinstance( @@ -390,41 +222,6 @@ def _compact_goal_boundary(boundary: dict[str, Any]) -> dict[str, Any]: return compact -def _compact_action_portfolio(portfolio: dict[str, Any]) -> dict[str, Any]: - compact = dict(portfolio) - suggested = portfolio.get("suggested_actions") - if isinstance(suggested, list): - compact["suggested_actions"] = [ - { - **{ - key: item[key] - for key in ("todo_id", "selection_role", "priority", "action_kind") - if key in item - }, - **{ - key: " ".join(str(item[key]).split())[:180] - for key in ("text", "continuation_hint") - if item.get(key) - }, - } - for item in suggested - if isinstance(item, dict) - ] - compact["suggested_action_details"] = { - "schema_version": QUOTA_CLI_ACTION_PORTFOLIO_COMPACTION_SCHEMA_VERSION, - "inlined_fields": [ - "todo_id", - "selection_role", - "priority", - "action_kind", - "text", - "continuation_hint", - ], - "ref": "$.agent_todo_summary.first_executable_items", - } - return compact - - def _compact_vision_continuation_audit( audit: dict[str, Any], ) -> dict[str, Any]: @@ -533,28 +330,6 @@ def _replace_nested_vision_continuation_audits( payload["interaction_contract"] = compact_interaction -def _compact_capability_gate(gate: dict[str, Any]) -> dict[str, Any]: - omitted_fields = ( - "available", - "candidate_order_policy", - "runnable_candidates", - "blocked_candidates", - "resolution_bindings", - ) - compact = {key: value for key, value in gate.items() if key not in omitted_fields} - compact["payload_compaction"] = { - "schema_version": QUOTA_CLI_CAPABILITY_GATE_COMPACTION_SCHEMA_VERSION, - "omitted_fields": [key for key in omitted_fields if key in gate], - "omitted_candidate_counts": { - "runnable": len(gate.get("runnable_candidates") or []), - "blocked": len(gate.get("blocked_candidates") or []), - "resolution_bindings": len(gate.get("resolution_bindings") or []), - }, - "full_detail_cold_path": QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND, - } - return compact - - def _deduplicate_next_action_warning( warning: dict[str, Any], *, @@ -768,6 +543,7 @@ def _reference_inline_goal_acceptance(payload: dict[str, Any]) -> dict[str, Any] def _reference_context( payload: dict[str, Any], *, task_title: bool = True, explore_guidance: bool = True, + diagnostics: dict[str, Any] | None = None, ) -> dict[str, Any]: """One coarse transport call for the existing typed context owner.""" contract = payload.get("interaction_contract") @@ -775,12 +551,51 @@ def _reference_context( context = agent.get("work_context") if isinstance(agent, dict) else None # No source carrier means neither operation can act. Leave semantic matching # and reference validation in TypeScript, rather than replicating its rules. - if not (task_title or explore_guidance) or not isinstance(context, dict) or not context.get("sources"): + if diagnostics is None and (not (task_title or explore_guidance) or not isinstance(context, dict) or not context.get("sources")): return payload from .turn_envelope import _turn_envelope_result - return _turn_envelope_result("reference_context", payload=payload, - task_title=task_title, explore_guidance=explore_guidance) + result = _turn_envelope_result("reference_context", payload=payload, + task_title=task_title, explore_guidance=explore_guidance, + **({"diagnostics": diagnostics} if diagnostics is not None else {})) + # Existing reference-only callers keep their historical wire contract. + return result if diagnostics is None else _apply_context_projection_changes(payload, result) + + +def _apply_context_projection_changes( + source: dict[str, Any], result: dict[str, Any], +) -> dict[str, Any]: + """Apply typed display edits to the original Python facts, without policy.""" + if set(result) != {"changes"} or not isinstance(result["changes"], list): + raise RuntimeError("TypeScript context projection must return changes") + # JSON transport has no aliasing. Keep Python's exact numeric values while + # separating repeated caller-owned objects before applying path edits. + projected = json.loads(json.dumps(source, allow_nan=False)) + try: + for change in result["changes"]: + path = change["path"] + if not isinstance(path, list) or not path: + raise ValueError("projection path must be nonempty") + parent: Any = projected + for key in path[:-1]: + parent = parent[key] + key = path[-1] + if "remove" in change: + del parent[key] + continue + if "from" in change: + value: Any = source + for part in change["from"]: + value = value[part] + else: + value = change["value"] + if isinstance(parent, list) and key == len(parent): + parent.append(deepcopy(value)) + else: + parent[key] = deepcopy(value) + except (KeyError, IndexError, TypeError, ValueError) as error: + raise RuntimeError("Invalid TypeScript context projection changes") from error + return projected def compact_quota_should_run_cli_payload( @@ -794,30 +609,6 @@ def compact_quota_should_run_cli_payload( """Bound CLI-only diagnostics after the full decision is computed.""" compact = payload - compacted_roles: list[str] = [] - summary = payload.get("agent_todo_summary") - if not include_todo_summary_detail and isinstance(summary, dict): - compact = dict(compact) - compact["agent_todo_summary"] = _compact_agent_todo_summary(summary) - compacted_roles.append("agent") - - user_summary = payload.get("user_todo_summary") - if not include_user_todo_summary_detail and isinstance(user_summary, dict): - compact = dict(compact) - compact["user_todo_summary"] = _compact_user_todo_summary(user_summary) - compacted_roles.append("user") - - if compacted_roles: - compact["todo_summary_projection"] = { - "schema_version": QUOTA_CLI_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION, - "mode": "compact_hot_path", - "compacted_roles": compacted_roles, - "detail_ref": ( - QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND - if compacted_roles[0] == "agent" - else QUOTA_CLI_USER_TODO_SUMMARY_DETAIL_COMMAND - ), - } goal_boundary = payload.get("goal_boundary") if not include_goal_boundary_detail and isinstance(goal_boundary, dict): compact_goal_boundary = _compact_goal_boundary(goal_boundary) @@ -843,16 +634,6 @@ def compact_quota_should_run_cli_payload( compact = dict(compact) compact["replan_action_packet"] = compact_replan_action if not include_todo_summary_detail: - action_portfolio = payload.get("action_portfolio") - if isinstance(action_portfolio, dict): - compact = dict(compact) - compact["action_portfolio"] = _compact_action_portfolio( - action_portfolio - ) - capability_gate = payload.get("capability_gate") - if isinstance(capability_gate, dict): - compact = dict(compact) - compact["capability_gate"] = _compact_capability_gate(capability_gate) warning = payload.get("next_action_projection_warning") if isinstance(warning, dict): compact = dict(compact) @@ -878,8 +659,13 @@ def compact_quota_should_run_cli_payload( agent_lane_next_action.pop("content_revision", None) compact["agent_lane_next_action"] = agent_lane_next_action compact = _reference_inline_goal_acceptance(compact) + diagnostics = None + if ((not include_todo_summary_detail and isinstance(payload.get("agent_todo_summary"), dict)) + or (not include_user_todo_summary_detail and isinstance(payload.get("user_todo_summary"), dict))): + diagnostics = {"agent_detail": include_todo_summary_detail, + "user_detail": include_user_todo_summary_detail} compact = _reference_context(compact, task_title=not include_todo_summary_detail, - explore_guidance=not include_vision_detail) + explore_guidance=not include_vision_detail, diagnostics=diagnostics) return _promote_interaction_contract( _promote_runtime_capability_reentry(compact) ) @@ -891,20 +677,9 @@ def compact_quota_plan_cli_payload( """Bound read-only CLI summaries after complete planning; retain full opt-ins.""" if payload.get("mode") not in {"status", "plan"} or not isinstance(payload.get("groups"), dict): return payload + if {"agent-todos", "user-todos"} <= detail_sections: + return payload - def project_row(row: dict[str, Any]) -> dict[str, Any]: - result = dict(row) - for role, compact_summary in (("agent", _compact_agent_todo_summary), ("user", _compact_user_todo_summary)): - key = f"{role}_todos" - if f"{role}-todos" not in detail_sections and isinstance(row.get(key), dict): - result[key] = compact_summary(row[key], detail_command=( - f"quota {payload['mode']} --include-detail {role}-todos" - )) - return result - - result = dict(payload) - result["groups"] = {state: [project_row(row) for row in rows] - for state, rows in payload["groups"].items()} - if isinstance(payload.get("next_automatic_turn"), dict): - result["next_automatic_turn"] = project_row(payload["next_automatic_turn"]) - return result + return _reference_context(payload, task_title=False, explore_guidance=False, diagnostics={ + "agent_detail": "agent-todos" in detail_sections, + "user_detail": "user-todos" in detail_sections}) diff --git a/loopx/control_plane/quota/diagnostic_projection.ts b/loopx/control_plane/quota/diagnostic_projection.ts new file mode 100644 index 0000000000..58d804ab60 --- /dev/null +++ b/loopx/control_plane/quota/diagnostic_projection.ts @@ -0,0 +1,169 @@ +/** Presentation of known Todo diagnostic lanes, after the decision is made. + * No selection, authority, task-text shortening or suffix-based field removal. + * Native callers provide the original capture reader; legacy callers retain + * their explicit current-state detail command, which cannot recover an older + * decision. This projection grants no admission or execution authority. + */ +import type {JsonObject} from "../effect_program.ts"; +import {jsonObject} from "../runtime_decode.ts"; +import {TODO_CANONICAL_READ_RECORD_FIELDS} from "../coordination/coordination_state_contract.ts"; + +type ProjectionPath = (string | number)[]; +type ProjectionChange = {path: ProjectionPath; value: unknown} | + {path: ProjectionPath; from: ProjectionPath} | {path: ProjectionPath; remove: true}; + +/** Return only display edits, never round-trip unrelated source facts through + * JavaScript numbers. References address the immutable input, including bodies + * moved into source_projection. Python applies these edits without policy. */ +export function contextProjectionChanges(source: JsonObject, projected: JsonObject): JsonObject { + const paths = new WeakMap(), changes: ProjectionChange[] = []; + const index = (value: unknown, path: ProjectionPath): void => { + if (value === null || typeof value !== "object" || paths.has(value)) return; + paths.set(value, path); + if (Array.isArray(value)) value.forEach((item, i) => index(item, [...path, i])); + else Object.entries(value).forEach(([key, item]) => index(item, [...path, key])); + }; + index(source, []); + const visit = (before: unknown, after: unknown, path: ProjectionPath): void => { + if (Object.is(before, after)) return; + const from = after !== null && typeof after === "object" ? paths.get(after) : undefined; + if (from) { changes.push({path, from}); return; } + if (Array.isArray(after)) { + const old = Array.isArray(before) ? before : []; + if (!Array.isArray(before)) changes.push({path, value: []}); + after.forEach((value, i) => visit(old[i], value, [...path, i])); + // Remove from the end so earlier indices retain their original meaning. + for (let i = old.length - 1; i >= after.length; i--) changes.push({path: [...path, i], remove: true}); + } else if (jsonObject(after)) { + const old = jsonObject(before) ?? {}; + if (!jsonObject(before)) changes.push({path, value: {}}); + for (const key of Object.keys(old)) + if (!Object.hasOwn(after as JsonObject, key)) changes.push({path: [...path, key], remove: true}); + for (const [key, value] of Object.entries(after as JsonObject)) + visit(Object.hasOwn(old, key) ? old[key] : undefined, value, [...path, key]); + } else changes.push({path, value: after}); + }; + visit(source, projected, []); + return {changes}; +} + +const AGENT_LIMITS: Readonly> = { + first_executable_items: 3, unclaimed_priority_open_items: 3, + monitor_due_items: 1, monitor_capability_blocked_due_items: 2, + monitor_schedule_gap_items: 1, current_agent_blocker_items: 2, +}; +const USER_LIMITS: Readonly> = { + first_open_items: 3, gate_open_items: 3, active_next_action_items: 3, +}; +// These are registered summary diagnostics, not a pattern over future lists. +const DIAGNOSTIC_LANES = new Set([ + "items", "first_open_items", "first_executable_items", "gate_open_items", + "monitor_open_items", "monitor_due_items", "watch_only_monitor_due_items", + "non_watch_only_monitor_due_items", "monitor_capability_blocked_due_items", + "monitor_schedule_gap_items", "active_next_action_items", "active_next_action_executable_items", + "backlog_items", "executable_backlog_items", "unclaimed_priority_open_items", + "claimed_open_items", "claimed_advancement_open_items", "claimed_monitor_open_items", + "current_agent_claimed_open_items", "current_agent_claimed_advancement_items", + "current_agent_claimed_monitor_items", "claimed_by_others_items", "current_agent_blocker_items", + "deferred_items", "deferred_resume_candidates", "current_agent_deferred_resume_candidates", + "unclaimed_deferred_resume_candidates", "other_agent_deferred_resume_candidates", + "executor_excluded_self_deferred_resume_candidates", "resume_blocked_items", "blocker_items", + "recent_completed_advancement_items", "completed_without_successor_items", + "other_agent_bound_user_action_items", +]); +// The read codec adds these display fields to canonical records. Retain them +// on visible items; they do not make a known diagnostic list a future schema. +const ITEM_FIELDS = new Set([...TODO_CANONICAL_READ_RECORD_FIELDS, "index", "content_revision"]); +function knownItems(items: unknown[]): boolean { + const scalar = (value: unknown) => value === null || ["string", "number", "boolean"].includes(typeof value); + return items.every(value => { + const item = jsonObject(value); + return item !== null && (item.schema_version === undefined || item.schema_version === "todo_item_v0") && + Object.entries(item).every(([key, field]) => ITEM_FIELDS.has(key) && + (scalar(field) || (Array.isArray(field) && field.every(scalar)))); + }); +} + +function compactSummary(summary: JsonObject, role: "agent" | "user", detail: string, captured: boolean): JsonObject { + if (summary.schema_version !== undefined && summary.schema_version !== "todo_summary_v0") return summary; + const limits = role === "agent" ? AGENT_LIMITS : USER_LIMITS; + const schema = role === "agent" ? "quota_cli_todo_summary_compaction_v0" : "quota_cli_user_todo_summary_compaction_v0"; + const compact = {...summary}, previous = jsonObject(summary.payload_compaction); + const owned = previous?.schema_version === schema; + const omitted: JsonObject = {...(owned ? jsonObject(previous?.omitted_lanes) : {})}; + for (const [key, value] of Object.entries(summary)) { + if (!DIAGNOSTIC_LANES.has(key) || !Array.isArray(value) || !knownItems(value)) continue; + const limit = limits[key] ?? 0; + if (limit) compact[key] = value.slice(0, limit); // Complete items, including unknown fields. + else delete compact[key]; + if (value.length > limit) omitted[key] = value.length - limit; + } + for (const [parentKey, fields] of [ + ["claim_scope", ["other_agent_claimed_items", "blocked_claimed_items"]], + ["todo_succession_warning", ["items"]], + ] as const) { + const parent = jsonObject(summary[parentKey]); + if (!parent) continue; + const projected = {...parent}; + for (const key of fields) { + const value = parent[key]; + if (!Array.isArray(value) || !knownItems(value)) continue; + if (parentKey === "todo_succession_warning" && !Object.hasOwn(parent, "todo_ids")) { + const ids = value.map(item => jsonObject(item)?.todo_id).filter(id => typeof id === "string" && id).slice(0, 3); + if (ids.length) projected.todo_ids = ids; + } + delete projected[key]; + if (value.length) omitted[`${parentKey}.${key}`] = value.length; + } + compact[parentKey] = projected; + } + const metadata: JsonObject = {...(owned ? previous : previous ? {source_projection: previous} : {}), + schema_version: schema, + retained_item_lanes: Object.keys(limits).filter(key => key !== "current_agent_blocker_items" || + (Array.isArray(summary[key]) && (summary[key] as unknown[]).length > 0)).sort(), + omitted_lanes: omitted}; + delete metadata.full_detail_cold_path; + delete metadata.full_detail_capture_ref; + metadata[captured ? "full_detail_capture_ref" : "full_detail_cold_path"] = detail; + compact.payload_compaction = metadata; + return compact; +} + +/** A coarse operation for all current callers, never a bridge call per row. */ +export function projectQuotaDiagnostics(payload: JsonObject, options: JsonObject = {}): JsonObject { + const agentDetail = options.agent_detail === true, userDetail = options.user_detail === true; + const captured = options.capture_ref === "$.source_capture"; + const command = (role: "agent" | "user", mode: string) => + captured ? "$.source_capture" : `quota ${mode} --include-detail ${role}-todos`; + if ((payload.mode === "status" || payload.mode === "plan") && jsonObject(payload.groups)) { + const row = (value: unknown): unknown => { + const source = jsonObject(value); + if (!source) return value; + const result = {...source}; + for (const role of ["agent", "user"] as const) { + const key = `${role}_todos`, summary = jsonObject(source[key]); + if (summary && !(role === "agent" ? agentDetail : userDetail)) + result[key] = compactSummary(summary, role, command(role, String(payload.mode)), captured); + } + return result; + }; + return {...payload, groups: Object.fromEntries(Object.entries(payload.groups as JsonObject) + .map(([state, rows]) => [state, Array.isArray(rows) ? rows.map(row) : rows])), + ...(Object.hasOwn(payload, "next_automatic_turn") ? {next_automatic_turn: row(payload.next_automatic_turn)} : {})}; + } + const result = {...payload}, roles: string[] = []; + for (const role of ["agent", "user"] as const) { + const key = `${role}_todo_summary`, summary = jsonObject(payload[key]); + if (!summary || (role === "agent" ? agentDetail : userDetail)) continue; + result[key] = compactSummary(summary, role, command(role, "should-run"), captured); + if (result[key] !== summary) roles.push(role); + } + if (roles.length) result.todo_summary_projection = { + schema_version: "quota_cli_todo_summary_compaction_v0", mode: "compact_hot_path", + compacted_roles: roles, detail_ref: command(roles[0] === "agent" ? "agent" : "user", "should-run"), + }; + // Candidate collections and portfolio text are decision context, not mere + // duplicate counters. Preserve them rather than relying on a fresh guard + // or a truncated summary to recover a different candidate's requirements. + return result; +} diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 98467d7d93..bfcf7c5c9d 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1,3 +1,4 @@ +import {contextProjectionChanges, projectQuotaDiagnostics} from "./diagnostic_projection.ts"; import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts"; import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts"; import { createHash } from "node:crypto"; @@ -15,7 +16,7 @@ import { } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { projectTurnStartUnavailableContext } from "../capability_hooks.ts"; -import { requireBoolean, requireJsonObject, requireNonEmptyString, requireStringLiteral } from "../runtime_decode.ts"; +import { jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString, requireStringLiteral } from "../runtime_decode.ts"; import { projectPendingCapabilityIntent } from "../work_items/pending_capability_intent.ts"; import { projectInteractionRequiredReads, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts"; import { measureTurnEnvelope, TURN_ENVELOPE_SECTION_TARGETS } from "./turn_envelope_budget.ts"; @@ -1455,10 +1456,12 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { return {ok: true}; } if (operation === "reference_context") { - let payload = requireJsonObject(request.payload, "quota decision"); + const source = requireJsonObject(request.payload, "quota decision"); + let payload = source; + if (jsonObject(request.diagnostics)) payload = projectQuotaDiagnostics(payload, request.diagnostics as JsonObject); if (request.task_title === true) payload = referenceCurrentTaskTitle(payload); if (request.explore_guidance === true) payload = referenceExploreGuidance(payload); - return payload; + return jsonObject(request.diagnostics) ? contextProjectionChanges(source, payload) : payload; } if (operation === "build") return buildTurnEnvelope(request); if (operation === "context_delivery") return prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); diff --git a/tests/control_plane/test_quota_action_portfolio.py b/tests/control_plane/test_quota_action_portfolio.py index b649089581..33c6e63b5c 100644 --- a/tests/control_plane/test_quota_action_portfolio.py +++ b/tests/control_plane/test_quota_action_portfolio.py @@ -261,38 +261,10 @@ def test_sticky_primary_exposes_bounded_agent_selection_on_every_hot_path() -> N assert compact["action_portfolio"]["selection_policy"] == ( portfolio["selection_policy"] ) - compact_suggestions = compact["action_portfolio"]["suggested_actions"] - assert compact_suggestions == [ - { - "todo_id": PRIMARY_ID, - "selection_role": "recommended", - "priority": "P0", - "action_kind": "monitor", - "text": "[P0] Run the Monday-only primary operation after its window opens.", - "continuation_hint": "Do not poll before the scheduled primary window.", - }, - { - "todo_id": FALLBACK_ID, - "selection_role": "alternative", - "priority": "P2", - "text": "[P2] Advance the independent fallback slice.", - "continuation_hint": ( - "Start from the independent fallback validation boundary." - ), - }, - ] - assert compact["action_portfolio"]["suggested_action_details"] == { - "schema_version": "quota_cli_action_portfolio_compaction_v1", - "inlined_fields": [ - "todo_id", - "selection_role", - "priority", - "action_kind", - "text", - "continuation_hint", - ], - "ref": "$.agent_todo_summary.first_executable_items", - } + # A selectable alternative's full requirements and text remain decision + # context, rather than a 180-character view with a partial summary pointer. + assert compact["action_portfolio"] == portfolio + assert compact["capability_gate"] == packet["capability_gate"] compact_candidates = compact["agent_todo_summary"]["first_executable_items"] assert compact_candidates[1]["required_capabilities"] == ["fallback_runner"] assert compact_candidates[1]["required_write_scopes"] == [ diff --git a/tests/control_plane/test_quota_cli_projection.py b/tests/control_plane/test_quota_cli_projection.py index 6fe13f08f6..09463544e1 100644 --- a/tests/control_plane/test_quota_cli_projection.py +++ b/tests/control_plane/test_quota_cli_projection.py @@ -351,15 +351,7 @@ def test_compact_quota_should_run_cli_payload_keeps_decision_lanes_and_counts() assert summary["open_count"] == 40 assert "first_open_items" not in summary assert len(summary["first_executable_items"]) == 3 - assert summary["first_executable_items"][0] == { - "schema_version": "todo_item_v0", - "todo_id": "execute-0", - "text": "execute item 0", - "status": "open", - "priority": "P1", - "task_class": "advancement_task", - "action_kind": "execute-action", - } + assert summary["first_executable_items"][0] == first_executable[0] assert [item["todo_id"] for item in summary["unclaimed_priority_open_items"]] == [ "unclaimed-0", "unclaimed-1", @@ -371,28 +363,7 @@ def test_compact_quota_should_run_cli_payload_keeps_decision_lanes_and_counts() "monitor-gap-0", ] assert summary["current_agent_blocker_count"] == 3 - assert summary["current_agent_blocker_items"] == [ - { - "schema_version": "todo_item_v0", - "todo_id": "blocker-0", - "text": "blocker item 0", - "status": "blocked", - "priority": "P1", - "task_class": "blocker", - "action_kind": "blocker-action", - "reason": "blocker reason 0", - }, - { - "schema_version": "todo_item_v0", - "todo_id": "blocker-1", - "text": "blocker item 1", - "status": "blocked", - "priority": "P1", - "task_class": "blocker", - "action_kind": "blocker-action", - "reason": "blocker reason 1", - }, - ] + assert summary["current_agent_blocker_items"] == current_agent_blockers[:2] assert "backlog_items" not in summary assert "claimed_open_items" not in summary assert "other_agent_claimed_items" not in summary["claim_scope"] @@ -764,7 +735,8 @@ def test_compact_quota_deduplicates_vision_without_losing_decision_rules() -> No ) assert len(json.dumps(payload, indent=2)) > 40_000 - assert len(json.dumps(compact, indent=2)) <= 40_000 + # The old fixture hid selectable candidates. Retain that context and measure the cost. + assert len(json.dumps(compact, indent=2)) < len(json.dumps(payload, indent=2)) assert compact["selected_todo"] == payload["selected_todo"] assert compact["interaction_contract"]["mode"] == "bounded_delivery" assert { @@ -784,7 +756,7 @@ def test_compact_quota_deduplicates_vision_without_losing_decision_rules() -> No assert compact["interaction_contract"]["agent_channel"][ "vision_continuation_audit" ]["vision_gap_judge"]["decision"] == "continue" - assert "runnable_candidates" not in compact["capability_gate"] + assert compact["capability_gate"] == payload["capability_gate"] assert compact["next_action_projection_warning"]["projection_refs"][ "active_state_next_action" ] == "$.active_state_next_action" @@ -998,3 +970,138 @@ def test_compact_quota_should_run_cli_payload_keeps_hot_path_identity_in_markdow "task_class=advancement_task action_kind=monitor-gap-action" ) in markdown assert "- interaction_contract: " in markdown + + +@pytest.mark.parametrize("role", ["agent", "user"]) +def test_shared_diagnostics_preserve_unknown_lists_nested_fields_and_complete_items(role): + lane = "first_executable_items" if role == "agent" else "first_open_items" + item = {**_items(1, prefix="retained")[0], "text": "Original constraint. " * 200, + "future_requirements": ["Keep the tail stop."], "lease": {"owner": "peer", "epoch": 7}} + unknown = [None, False, 0, {"requirement": "A new obligation"}] + summary = {"schema_version": "todo_summary_v0", lane: [item], "backlog_items": _items(5, prefix="diagnostic"), + "future_items": unknown, "future_rules": unknown, + "deferred_items": [{"todo_id": "later", "future_global_stop": "Do not continue."}], + "recent_completed_advancement_items": [{"todo_id": "old", "completion_result": {"future_stop": True}}], + "claim_scope": {"owner": "peer", "future_items": unknown, "blocked_claimed_items": []}, + "future_object": {"items": unknown}, + "todo_succession_warning": {"items": _items(4, prefix="old"), "todo_ids": ["original-id"], "future_items": unknown}} + payload = {f"{role}_todo_summary": summary, "capability_gate": {"runnable_candidates": [item], "future": unknown}, + "action_portfolio": {"suggested_actions": [item]}, "planning_horizon": {"work_items": [item]}} + original = deepcopy(payload) + projected = compact_quota_should_run_cli_payload(payload) + result = projected[f"{role}_todo_summary"] + assert result[lane] == [item] + assert result["future_items"] == result["future_rules"] == unknown + assert result["future_object"] == {"items": unknown} + assert result["deferred_items"] == summary["deferred_items"] + assert result["recent_completed_advancement_items"] == summary["recent_completed_advancement_items"] + assert result["claim_scope"] == {"owner": "peer", "future_items": unknown} + assert result["todo_succession_warning"]["todo_ids"] == ["original-id"] + assert result["todo_succession_warning"]["future_items"] == unknown + for key in ("capability_gate", "action_portfolio", "planning_horizon"): + assert projected[key] == payload[key] + assert payload == original + assert compact_quota_should_run_cli_payload(projected) == projected + detailed = compact_quota_should_run_cli_payload(payload, + include_todo_summary_detail=True, include_user_todo_summary_detail=True) + assert detailed == payload + + +def test_readonly_plan_projects_all_rows_in_one_typed_call(monkeypatch): + from loopx.control_plane.quota.cli_projection import compact_quota_plan_cli_payload + from loopx.control_plane.quota import turn_envelope + summary = {"schema_version": "todo_summary_v0", "backlog_items": _items(3, prefix="cold"), + "first_executable_items": _items(1, prefix="work"), "future_items": [False, None]} + row = {"agent_todos": summary, "user_todos": summary} + calls, real = [], turn_envelope._turn_envelope_result + def observed(operation, **params): + calls.append(operation) + return real(operation, **params) + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", observed) + payload = {"mode": "plan", "groups": {"ready": [row] * 8}, "next_automatic_turn": row} + projected = compact_quota_plan_cli_payload(payload, detail_sections=frozenset({"user-todos"})) + assert calls == ["reference_context"] + for result in [*projected["groups"]["ready"], projected["next_automatic_turn"]]: + assert result["user_todos"] == summary + assert "backlog_items" not in result["agent_todos"] + assert result["agent_todos"]["future_items"] == [False, None] + assert result["agent_todos"]["payload_compaction"]["full_detail_cold_path"] == "quota plan --include-detail agent-todos" + + +@pytest.mark.parametrize("mode", ["should-run", "status", "plan"]) +@pytest.mark.parametrize("expanded", [None, "agent", "user"]) +def test_diagnostic_transport_preserves_source_number_spelling_and_precision(mode, expanded): + from loopx.control_plane.quota.cli_projection import compact_quota_plan_cli_payload + + facts = {"compute": 1.0, "negative_zero": -0.0, "precise": 2**60 + 1, + "nested": [None, False, 0, {"value": 3.0, "tail": "Keep this final stop."}]} + summary = {"schema_version": "todo_summary_v0", "backlog_items": _items(5, prefix="cold"), + "future_items": [facts], "first_open_items": _items(4, prefix="user"), + "first_executable_items": _items(4, prefix="agent"), + "payload_compaction": {"schema_version": "upstream_v1", "facts": facts}} + row = {"quota": facts, "agent_todos": summary, "user_todos": summary} + payload = ({"mode": mode, "quota": facts, "agent_todo_summary": summary, + "user_todo_summary": summary} if mode == "should-run" else + {"mode": mode, "groups": {"eligible": [row]}, "next_automatic_turn": row}) + encoded = json.dumps(payload, ensure_ascii=False) + + def project(value): + if mode == "should-run": + return compact_quota_should_run_cli_payload(value, + include_todo_summary_detail=expanded == "agent", + include_user_todo_summary_detail=expanded == "user") + return compact_quota_plan_cli_payload(value, + detail_sections=frozenset({f"{expanded}-todos"}) if expanded else frozenset()) + + result = project(payload) + rows = [result] if mode == "should-run" else [result["groups"]["eligible"][0], result["next_automatic_turn"]] + for actual in rows: + assert json.dumps(actual["quota"], ensure_ascii=False) == json.dumps(facts, ensure_ascii=False) + for role in ("agent", "user"): + value = actual[f"{role}_todo_summary" if mode == "should-run" else f"{role}_todos"] + assert json.dumps(value["future_items"], ensure_ascii=False) == json.dumps([facts], ensure_ascii=False) + if role == expanded: + assert json.dumps(value, ensure_ascii=False) == json.dumps(summary, ensure_ascii=False) + else: + assert "backlog_items" not in value + assert json.dumps(value["payload_compaction"]["source_projection"], ensure_ascii=False) == json.dumps(summary["payload_compaction"], ensure_ascii=False) + assert json.dumps(payload, ensure_ascii=False) == encoded + assert json.dumps(project(result), ensure_ascii=False) == json.dumps(result, ensure_ascii=False) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("mode", ["status", "plan"]) +def test_public_readonly_projection_preserves_canonical_work_and_detail(tmp_path, monkeypatch, provider, mode): + from tests.control_plane import test_quota_settlement_cli as fixture + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = fixture._write_fixture(tmp_path) + state = fixture._configure_read_only_todo(project) + fixture._initialize_fixture_authority(project, runtime, provider=provider) + def canonical_work(): + rc, value = fixture._run_cli(registry, runtime, "todo", "list", "--goal-id", fixture.GOAL_ID, + "--todo-id", fixture.TODO_ID, cwd=project) + assert rc == 0 and value["matched"] is True, value + assert value["authority_read"]["source_authority"] == f"{provider}_v0" + assert value["authority_read"]["legacy_fallback_used"] is False + return value["todo"], value["authority_read"]["provider_revision"] + before = canonical_work() + original = state.read_bytes() + def call(*detail): + rc, value = fixture._run_cli(registry, runtime, "quota", mode, *detail, cwd=project) + assert rc == 0 and value["ok"] is True, value + return {row["goal_id"]: row for rows in value["groups"].values() for row in rows} + projected = call()[fixture.GOAL_ID] + detailed = call("--include-detail", "agent-todos", "--include-detail", "user-todos")[fixture.GOAL_ID] + for role in ("agent", "user"): + one, full = projected[f"{role}_todos"], detailed[f"{role}_todos"] + assert one["open_count"] == full["open_count"] + assert one["work_counts"] == full["work_counts"] + assert one["payload_compaction"]["full_detail_cold_path"] == f"quota {mode} --include-detail {role}-todos" + lane = "first_executable_items" if role == "agent" else "first_open_items" + assert one[lane] == full[lane] + assert canonical_work() == before + assert state.read_bytes() == original + assert fixture._spend_run_count(runtime) == 0 + assert fixture._heartbeat_receipt_count(runtime, "readonly-projection") == 0 diff --git a/tests/control_plane/test_quota_context_reference_owner.py b/tests/control_plane/test_quota_context_reference_owner.py index cd9fbbadec..76ef472719 100644 --- a/tests/control_plane/test_quota_context_reference_owner.py +++ b/tests/control_plane/test_quota_context_reference_owner.py @@ -158,7 +158,9 @@ def observed(operation, **params): task_detail, vision_detail = detail in {"task", "both"}, detail in {"vision", "both"} result = compact_quota_should_run_cli_payload(source, include_todo_summary_detail=task_detail, include_vision_detail=vision_detail) - assert len(calls) == (0 if detail == "both" else 1) + # Agent/vision detail leaves the User summary on its default compact path. + # That formerly local Python work now shares this single typed request. + assert calls == ["reference_context"] if task_detail: assert result["agent_lane_next_action"]["title"] == title else: @@ -171,6 +173,18 @@ def observed(operation, **params): assert result["interaction_contract"]["cli_channel"] == source["interaction_contract"]["cli_channel"] +def test_all_explicit_details_need_no_diagnostic_transport(monkeypatch): + from loopx.control_plane.quota.cli_projection import compact_quota_should_run_cli_payload + from loopx.control_plane.quota import turn_envelope + source, _ = packet() + def unavailable(*args, **kwargs): + raise AssertionError("explicit full summaries and guidance need no projection") + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", unavailable) + result = compact_quota_should_run_cli_payload(source, include_todo_summary_detail=True, + include_user_todo_summary_detail=True, include_vision_detail=True) + assert result == source + + def test_no_context_sources_needs_no_transport(monkeypatch): from loopx.control_plane.quota import turn_envelope def unavailable(*args, **kwargs):