From b5bd43bf176912758b954ae222bb225553b168fd Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 29 Oct 2026 10:16:45 +0800 Subject: [PATCH 1/3] fix(control-plane): bound large quota planning requests Signed-off-by: huangruiteng --- loopx/control_plane/effect_runtime.py | 1 + loopx/control_plane/effect_runtime_server.ts | 1 + loopx/control_plane/todos/quota_selection.py | 20 +++-- loopx/control_plane/todos/resume_planning.py | 90 +++++++++++++++++++ loopx/control_plane/todos/resume_planning.ts | 25 +++++- tests/control_plane/test_quota_selection.py | 63 +++++++++++++ .../test_resume_planning_projection.py | 30 ++++++- 7 files changed, 218 insertions(+), 12 deletions(-) diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 907a9b2fab..3c3c3625b6 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -43,6 +43,7 @@ MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024 LOCAL_SNAPSHOT_METHODS = frozenset({ "todo.read", + "todo.quota_planning.project", "collaboration.inbox.read", "collaboration.delegation.inventory_page", "benchmark.trajectory.codex-usage", diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index a1336023e4..651f9c0f3b 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -29,6 +29,7 @@ const MAX_INLINE_RESPONSE_BYTES = 2 * 1024 * 1024; // Explicit opt-in: ordinary effects retain the 2 MiB request/response wire. const LOCAL_SNAPSHOT_METHODS = new Set([ "todo.read", + "todo.quota_planning.project", "collaboration.inbox.read", "collaboration.delegation.inventory_page", "benchmark.trajectory.codex-usage", diff --git a/loopx/control_plane/todos/quota_selection.py b/loopx/control_plane/todos/quota_selection.py index 2553f51f81..f5b87f345e 100644 --- a/loopx/control_plane/todos/quota_selection.py +++ b/loopx/control_plane/todos/quota_selection.py @@ -21,7 +21,10 @@ todo_item_task_class, todo_projection_sort_key, todo_summary_monitor_writeback_supported, ) -from .resume_planning import build_todo_resume_planning_request +from .resume_planning import ( + build_referenced_todo_resume_planning_request, + rehydrate_todo_resume_planning_result, +) from .handoff_note import todo_read_source from .user_gate import is_user_gate_todo_item from .route_continuation import build_todo_route_continuation_facts @@ -151,6 +154,12 @@ def active(key: str) -> list[dict[str, Any]]: raw = value.get(key) return [encode(item) for item in raw if isinstance(item, dict)] if isinstance(raw, list) else [] + resume_request, resume_payloads = build_referenced_todo_resume_planning_request( + value, + agent_id=agent, + item_limit=8, + available_capabilities=(available_capabilities or []) if resolve_capacity else None, + ) try: wire = effect_runtime_result("todo.quota_planning.project", { "schema_version": "todo_quota_planning_request_v4", @@ -161,8 +170,7 @@ def active(key: str) -> list[dict[str, Any]]: "excluded": normalize_todo_excluded_agents(gate.get("excluded_agents"))} for gate in handoff_gates], "source_contract": _closure_source_facts(value), - "resume": build_todo_resume_planning_request(value, agent_id=agent, item_limit=8, - available_capabilities=(available_capabilities or []) if resolve_capacity else None), + "resume": resume_request, "selection": { "observed_at": observed_at, "available": normalize_required_capabilities(available_capabilities), @@ -177,10 +185,12 @@ def active(key: str) -> list[dict[str, Any]]: "frontier_revision_index": value.get("advancement_frontier_revision_index"), "diagnostic_limit": 3, "backlog_limit": 8, "visibility_limit": 16, }, - }) + }, large_local_snapshot=True) except EffectRuntimeRejected as exc: raise ValueError(str(exc)) from None - result = _decode_quota_planning_wire(wire) + result = rehydrate_todo_resume_planning_result( + _decode_quota_planning_wire(wire), resume_payloads + ) if not isinstance(result, dict) or result.get("schema_version") != "todo_quota_planning_v0": raise RuntimeError("TypeScript Todo quota planning shape mismatch") if not isinstance(result.get("lanes"), dict) or not isinstance(result["lanes"].get("gate_items"), list): diff --git a/loopx/control_plane/todos/resume_planning.py b/loopx/control_plane/todos/resume_planning.py index ca45138a47..0b47ec7cdd 100644 --- a/loopx/control_plane/todos/resume_planning.py +++ b/loopx/control_plane/todos/resume_planning.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from typing import Any from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result @@ -20,6 +21,19 @@ "current_agent_claimed_monitor_items", "claimed_monitor_open_items", ) +_REFERENCE_REQUEST_SCHEMA = "todo_resume_planning_request_v1" +_PAYLOAD_REFERENCE_KEY = "_loopx_resume_record_ref" +# These are the only payload facts read by resume_planning.ts and the shared +# resume-condition evaluator. Full compact payloads remain in Python and are +# restored after the typed projection returns. +_REFERENCE_PAYLOAD_FIELDS = ( + "index", "text", "todo_id", "role", "status", "archive_state", + "source_section", "task_class", "claimed_by", "task_repository", + "next_due_at", "expires_at", "resume_when", "resume_monitor_generation", + "material_change_generation", "resume_condition", "resume_ready", + "blocked_retry", +) + def _planning_item(item: dict[str, Any]) -> dict[str, Any]: payload = compact_todo_projection_item( @@ -83,6 +97,82 @@ def build_todo_resume_planning_request( } +def build_referenced_todo_resume_planning_request( + value: Any, *, agent_id: str | None = None, item_limit: int = 5, + available_capabilities: Any = None, +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + """Encode one semantic snapshot without repeating full payloads per lane. + + The ordinary v0 builder remains the compatibility contract for callers that + inspect or rewrite its source rows. Quota planning uses this referenced v1 + wire and rehydrates the typed result from the returned payload table. + """ + request = build_todo_resume_planning_request( + value, + agent_id=agent_id, + item_limit=item_limit, + available_capabilities=available_capabilities, + ) + records: list[dict[str, Any]] = [] + payloads: list[dict[str, Any]] = [] + references: dict[str, int] = {} + sources: dict[str, list[int]] = {} + for source_name, source_rows in request["sources"].items(): + source_refs: list[int] = [] + for row in source_rows: + # Equal normalized records are semantically interchangeable. Use + # their exact JSON form instead of Todo id alone so conflicting + # historical facts remain separate and are still diagnosed. + identity = json.dumps( + row, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + reference = references.get(identity) + if reference is None: + reference = len(records) + references[identity] = reference + payload = dict(row.get("payload") or {}) + payloads.append(payload) + wire_payload = { + key: payload[key] + for key in _REFERENCE_PAYLOAD_FIELDS + if key in payload + } + wire_payload[_PAYLOAD_REFERENCE_KEY] = reference + records.append({**row, "payload": wire_payload}) + source_refs.append(reference) + sources[source_name] = source_refs + return ( + { + **request, + "schema_version": _REFERENCE_REQUEST_SCHEMA, + "sources": sources, + "records": records, + }, + payloads, + ) + + +def rehydrate_todo_resume_planning_result( + value: Any, payloads: list[dict[str, Any]], +) -> Any: + """Replace private wire references with their complete compact payloads.""" + if isinstance(value, list): + return [rehydrate_todo_resume_planning_result(item, payloads) for item in value] + if not isinstance(value, dict): + return value + projected = { + key: rehydrate_todo_resume_planning_result(item, payloads) + for key, item in value.items() + if key != _PAYLOAD_REFERENCE_KEY + } + reference = value.get(_PAYLOAD_REFERENCE_KEY) + if reference is None: + return projected + if type(reference) is not int or not 0 <= reference < len(payloads): + raise RuntimeError("TypeScript Todo resume planning record reference is invalid") + return {**payloads[reference], **projected} + + def project_todo_resume_planning( value: Any, *, agent_id: str | None = None, item_limit: int = 5, available_capabilities: Any = None, diff --git a/loopx/control_plane/todos/resume_planning.ts b/loopx/control_plane/todos/resume_planning.ts index 143942450b..993e1902a9 100644 --- a/loopx/control_plane/todos/resume_planning.ts +++ b/loopx/control_plane/todos/resume_planning.ts @@ -12,6 +12,7 @@ import { } from "./resume_condition.ts"; export const RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v0"; +export const REFERENCED_RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v1"; export const RESUME_PLANNING_RESULT = "todo_resume_planning_v0"; const DEFERRED_POLICY = "quota may wake the current peer only for ready deferred todos " + "claimed by that agent or unclaimed; other-agent deferred todos remain " + @@ -186,12 +187,18 @@ function resolveCapacity(items: readonly Item[], source: readonly Item[], capabi }); } -function decodeSources(value: unknown): Record { +function decodeSources(value: unknown, records: Item[] | null = null): Record { const rawSources = requireJsonObject(value, "sources"); const sources = {} as Record; for (const key of SOURCE_KEYS) { if (!Array.isArray(rawSources[key])) throw new EffectRuntimeRequestError(`${key} must be an array`); - sources[key] = rawSources[key].map(item); + sources[key] = records === null ? rawSources[key].map(item) : rawSources[key].map((value) => { + const reference = requireInteger(value, `${key} record reference`); + if (reference < 0 || reference >= records.length) { + throw new EffectRuntimeRequestError(`${key} record reference is out of range`); + } + return records[reference]; + }); } return sources; } @@ -295,10 +302,20 @@ function blockedVisibility(resumeBlocked: Item[], monitorItems: Item[], options: * No mutation, claim, lease grant, monitor poll, or recovery effect is emitted. */ export function projectTodoResumePlanning(value: unknown): JsonObject { const request = requireJsonObject(value, "resume planning request"); - if (request.schema_version !== RESUME_PLANNING_REQUEST) { + if (![RESUME_PLANNING_REQUEST, REFERENCED_RESUME_PLANNING_REQUEST].includes( + String(request.schema_version), + )) { throw new EffectRuntimeRequestError("resume planning schema mismatch"); } - const sources = decodeSources(request.sources); + const records = request.schema_version === REFERENCED_RESUME_PLANNING_REQUEST + ? (() => { + if (!Array.isArray(request.records)) { + throw new EffectRuntimeRequestError("resume planning records must be an array"); + } + return request.records.map(item); + })() + : null; + const sources = decodeSources(request.sources, records); diagnoseSources(sources); const display: DisplayOptions = { agent: optionalNonEmptyString(request.agent_id, "agent_id"), diff --git a/tests/control_plane/test_quota_selection.py b/tests/control_plane/test_quota_selection.py index a07ba4dcbd..c6425d7865 100644 --- a/tests/control_plane/test_quota_selection.py +++ b/tests/control_plane/test_quota_selection.py @@ -58,6 +58,69 @@ def test_agent_execution_still_respects_claim_and_exclusion(): assert result["claim_scope"]["other_agent_claimed_open_count"] == 1 +@pytest.mark.parametrize("gate_scope", [False, True]) +def test_large_quota_history_uses_snapshot_without_losing_scope( + monkeypatch, gate_scope, +): + from loopx.control_plane import effect_runtime + + observed_sizes = [] + original = effect_runtime._request_with_info + + def observe(info, **kwargs): + if kwargs["method"] == "todo.quota_planning.project": + observed_sizes.append(len(json.dumps( + kwargs["params"], separators=(",", ":") + ).encode())) + return original(info, **kwargs) + + monkeypatch.setattr(effect_runtime, "_request_with_info", observe) + history = [ + item( + f"todo_history_{index}", + status="done", + task_class="advancement_task", + note="Synthetic historical evidence. " * 180, + index=index + 1, + ) + for index in range(2500) + ] + tail = [ + item("todo_tail_gate", blocks_agent="agent-b", index=2501), + item( + "todo_peer", task_class="advancement_task", + claimed_by="agent-a", index=2502, + ), + item( + "todo_owned", task_class="advancement_task", + claimed_by="agent-b", index=2503, + ), + ] + source = summary(history + tail) + source.update( + done_count=len(history), open_count=len(tail), first_open_items=tail, + ) + + result = summarize_user_todos_for_quota( + source, + agent_identity={"agent_id": "agent-b"}, + filter_user_gate_blocks_agent=gate_scope, + ) + + # Unique historical payloads cannot be reduced below the inline wire limit + # without dropping facts. Use the existing digest-verified local snapshot + # transport rather than raising or truncating. + assert observed_sizes and max(observed_sizes) > effect_runtime.MAX_REQUEST_BYTES + assert [row["todo_id"] for row in result["gate_open_items"]] == [ + "todo_tail_gate" + ] + assert [row["todo_id"] for row in result["first_executable_items"]] == ( + [] if gate_scope else ["todo_owned"] + ) + if not gate_scope: + assert result["claim_scope"]["other_agent_claimed_open_count"] == 1 + + @pytest.mark.parametrize("agent", [None, "agent-b"]) def test_route_replan_visibility_counts_before_budget_without_granting_execution(agent): own = [item(f"todo_route_{i}", task_class="advancement_task", claimed_by="agent-b", diff --git a/tests/control_plane/test_resume_planning_projection.py b/tests/control_plane/test_resume_planning_projection.py index 077fc15499..3a3e585e67 100644 --- a/tests/control_plane/test_resume_planning_projection.py +++ b/tests/control_plane/test_resume_planning_projection.py @@ -11,7 +11,10 @@ from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos -from loopx.control_plane.todos.resume_planning import project_todo_resume_planning +from loopx.control_plane.todos.resume_planning import ( + build_referenced_todo_resume_planning_request, + project_todo_resume_planning, +) from loopx.control_plane.todos.quota_summary import select_quota_todo_summary @@ -53,6 +56,27 @@ def ids(items: list[dict]) -> list[str]: return [item["todo_id"] for item in items] +def test_referenced_request_deduplicates_lanes_and_keeps_full_payload_local() -> None: + row = waiting( + "todo_shared", + content_revision="sha256:synthetic-full-projection", + ) + request, payloads = build_referenced_todo_resume_planning_request({ + "items": [row], + "first_open_items": [row], + "deferred_items": [row], + }) + + assert request["schema_version"] == "todo_resume_planning_request_v1" + assert len(request["records"]) == 1 + assert request["sources"]["items"] == [0] + assert request["sources"]["first_open_items"] == [0] + assert request["sources"]["deferred_items"] == [0] + assert request["records"][0]["payload"]["_loopx_resume_record_ref"] == 0 + assert "content_revision" not in request["records"][0]["payload"] + assert payloads[0]["content_revision"] == row["content_revision"] + + def test_claim_priority_and_exclusion_are_independent_of_display_priority() -> None: items = [waiting("todo_unclaimed", priority="P0"), waiting("todo_current", claimed_by="agent-a", priority="P4"), @@ -239,9 +263,9 @@ def test_quota_composes_capacity_and_visibility_in_one_request_per_source(monkey original = quota_selection.effect_runtime_result calls = [] - def record(method, params): + def record(method, params, **kwargs): calls.append(method) - return original(method, params) + return original(method, params, **kwargs) monkeypatch.setattr(quota_selection, "effect_runtime_result", record) summary = select_quota_todo_summary( From 35b6d95f7ffa909f85faab16605250ac31392f27 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 29 Oct 2026 15:11:55 +0800 Subject: [PATCH 2/3] fix(control-plane): preserve complete large-goal planning records Signed-off-by: huangruiteng --- loopx/control_plane/effect_runtime.py | 1 + loopx/control_plane/effect_runtime_server.ts | 1 + loopx/control_plane/goals/goal_vision_wait.py | 6 +- loopx/control_plane/todos/quota_selection.py | 11 +-- loopx/control_plane/todos/resume_planning.py | 60 ++-------------- loopx/control_plane/todos/resume_planning.ts | 5 +- tests/control_plane/test_quota_selection.py | 70 +++++++++++++++++++ .../test_resume_planning_projection.py | 37 ++++++++-- .../test_vision_wait_coverage.py | 4 +- 9 files changed, 120 insertions(+), 75 deletions(-) diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 3c3c3625b6..2c999f4360 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -44,6 +44,7 @@ LOCAL_SNAPSHOT_METHODS = frozenset({ "todo.read", "todo.quota_planning.project", + "goal.vision_wait.project", "collaboration.inbox.read", "collaboration.delegation.inventory_page", "benchmark.trajectory.codex-usage", diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 651f9c0f3b..e8a0de86e8 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -30,6 +30,7 @@ const MAX_INLINE_RESPONSE_BYTES = 2 * 1024 * 1024; const LOCAL_SNAPSHOT_METHODS = new Set([ "todo.read", "todo.quota_planning.project", + "goal.vision_wait.project", "collaboration.inbox.read", "collaboration.delegation.inventory_page", "benchmark.trajectory.codex-usage", diff --git a/loopx/control_plane/goals/goal_vision_wait.py b/loopx/control_plane/goals/goal_vision_wait.py index afb680418b..0796306709 100644 --- a/loopx/control_plane/goals/goal_vision_wait.py +++ b/loopx/control_plane/goals/goal_vision_wait.py @@ -6,7 +6,7 @@ from ..effect_runtime import effect_runtime_result from ..todos.contract import normalize_todo_id -from ..todos.resume_planning import build_todo_resume_planning_request +from ..todos.resume_planning import build_referenced_todo_resume_planning_request GOAL_VISION_WAIT_STATE_SCHEMA_VERSION = "goal_vision_wait_state_v0" @@ -100,8 +100,8 @@ def build_goal_vision_wait_state( "acceptance_gaps": acceptance_gaps or [], "selectable_advancement_count": selectable_advancement_count, "lineage_source_items": _vision_wait_lineage_rows(lineage_items), - "resume_planning_request": build_todo_resume_planning_request(planning_input, agent_id=agent_id), - }) + "resume_planning_request": build_referenced_todo_resume_planning_request(planning_input, agent_id=agent_id), + }, large_local_snapshot=True) if result is not None and ( not isinstance(result, dict) or result.get("schema_version") != GOAL_VISION_WAIT_STATE_SCHEMA_VERSION diff --git a/loopx/control_plane/todos/quota_selection.py b/loopx/control_plane/todos/quota_selection.py index f5b87f345e..eb1608fb4a 100644 --- a/loopx/control_plane/todos/quota_selection.py +++ b/loopx/control_plane/todos/quota_selection.py @@ -21,10 +21,7 @@ todo_item_task_class, todo_projection_sort_key, todo_summary_monitor_writeback_supported, ) -from .resume_planning import ( - build_referenced_todo_resume_planning_request, - rehydrate_todo_resume_planning_result, -) +from .resume_planning import build_referenced_todo_resume_planning_request from .handoff_note import todo_read_source from .user_gate import is_user_gate_todo_item from .route_continuation import build_todo_route_continuation_facts @@ -154,7 +151,7 @@ def active(key: str) -> list[dict[str, Any]]: raw = value.get(key) return [encode(item) for item in raw if isinstance(item, dict)] if isinstance(raw, list) else [] - resume_request, resume_payloads = build_referenced_todo_resume_planning_request( + resume_request = build_referenced_todo_resume_planning_request( value, agent_id=agent, item_limit=8, @@ -188,9 +185,7 @@ def active(key: str) -> list[dict[str, Any]]: }, large_local_snapshot=True) except EffectRuntimeRejected as exc: raise ValueError(str(exc)) from None - result = rehydrate_todo_resume_planning_result( - _decode_quota_planning_wire(wire), resume_payloads - ) + result = _decode_quota_planning_wire(wire) if not isinstance(result, dict) or result.get("schema_version") != "todo_quota_planning_v0": raise RuntimeError("TypeScript Todo quota planning shape mismatch") if not isinstance(result.get("lanes"), dict) or not isinstance(result["lanes"].get("gate_items"), list): diff --git a/loopx/control_plane/todos/resume_planning.py b/loopx/control_plane/todos/resume_planning.py index 0b47ec7cdd..24a5da4f0a 100644 --- a/loopx/control_plane/todos/resume_planning.py +++ b/loopx/control_plane/todos/resume_planning.py @@ -22,17 +22,6 @@ ) _REFERENCE_REQUEST_SCHEMA = "todo_resume_planning_request_v1" -_PAYLOAD_REFERENCE_KEY = "_loopx_resume_record_ref" -# These are the only payload facts read by resume_planning.ts and the shared -# resume-condition evaluator. Full compact payloads remain in Python and are -# restored after the typed projection returns. -_REFERENCE_PAYLOAD_FIELDS = ( - "index", "text", "todo_id", "role", "status", "archive_state", - "source_section", "task_class", "claimed_by", "task_repository", - "next_due_at", "expires_at", "resume_when", "resume_monitor_generation", - "material_change_generation", "resume_condition", "resume_ready", - "blocked_retry", -) def _planning_item(item: dict[str, Any]) -> dict[str, Any]: @@ -100,12 +89,12 @@ def build_todo_resume_planning_request( def build_referenced_todo_resume_planning_request( value: Any, *, agent_id: str | None = None, item_limit: int = 5, available_capabilities: Any = None, -) -> tuple[dict[str, Any], list[dict[str, Any]]]: +) -> dict[str, Any]: """Encode one semantic snapshot without repeating full payloads per lane. The ordinary v0 builder remains the compatibility contract for callers that - inspect or rewrite its source rows. Quota planning uses this referenced v1 - wire and rehydrates the typed result from the returned payload table. + inspect or rewrite its source rows. The v1 wire changes only repeated lane + bodies: every record retains its complete v0 facts for all typed consumers. """ request = build_todo_resume_planning_request( value, @@ -114,7 +103,6 @@ def build_referenced_todo_resume_planning_request( available_capabilities=available_capabilities, ) records: list[dict[str, Any]] = [] - payloads: list[dict[str, Any]] = [] references: dict[str, int] = {} sources: dict[str, list[int]] = {} for source_name, source_rows in request["sources"].items(): @@ -130,47 +118,11 @@ def build_referenced_todo_resume_planning_request( if reference is None: reference = len(records) references[identity] = reference - payload = dict(row.get("payload") or {}) - payloads.append(payload) - wire_payload = { - key: payload[key] - for key in _REFERENCE_PAYLOAD_FIELDS - if key in payload - } - wire_payload[_PAYLOAD_REFERENCE_KEY] = reference - records.append({**row, "payload": wire_payload}) + records.append(row) source_refs.append(reference) sources[source_name] = source_refs - return ( - { - **request, - "schema_version": _REFERENCE_REQUEST_SCHEMA, - "sources": sources, - "records": records, - }, - payloads, - ) - - -def rehydrate_todo_resume_planning_result( - value: Any, payloads: list[dict[str, Any]], -) -> Any: - """Replace private wire references with their complete compact payloads.""" - if isinstance(value, list): - return [rehydrate_todo_resume_planning_result(item, payloads) for item in value] - if not isinstance(value, dict): - return value - projected = { - key: rehydrate_todo_resume_planning_result(item, payloads) - for key, item in value.items() - if key != _PAYLOAD_REFERENCE_KEY - } - reference = value.get(_PAYLOAD_REFERENCE_KEY) - if reference is None: - return projected - if type(reference) is not int or not 0 <= reference < len(payloads): - raise RuntimeError("TypeScript Todo resume planning record reference is invalid") - return {**payloads[reference], **projected} + return {**request, "schema_version": _REFERENCE_REQUEST_SCHEMA, + "sources": sources, "records": records} def project_todo_resume_planning( diff --git a/loopx/control_plane/todos/resume_planning.ts b/loopx/control_plane/todos/resume_planning.ts index 993e1902a9..12bb62ff2e 100644 --- a/loopx/control_plane/todos/resume_planning.ts +++ b/loopx/control_plane/todos/resume_planning.ts @@ -302,9 +302,8 @@ function blockedVisibility(resumeBlocked: Item[], monitorItems: Item[], options: * No mutation, claim, lease grant, monitor poll, or recovery effect is emitted. */ export function projectTodoResumePlanning(value: unknown): JsonObject { const request = requireJsonObject(value, "resume planning request"); - if (![RESUME_PLANNING_REQUEST, REFERENCED_RESUME_PLANNING_REQUEST].includes( - String(request.schema_version), - )) { + if (request.schema_version !== RESUME_PLANNING_REQUEST && + request.schema_version !== REFERENCED_RESUME_PLANNING_REQUEST) { throw new EffectRuntimeRequestError("resume planning schema mismatch"); } const records = request.schema_version === REFERENCED_RESUME_PLANNING_REQUEST diff --git a/tests/control_plane/test_quota_selection.py b/tests/control_plane/test_quota_selection.py index c6425d7865..c33cee9cbf 100644 --- a/tests/control_plane/test_quota_selection.py +++ b/tests/control_plane/test_quota_selection.py @@ -1,6 +1,7 @@ """Quota scope is distinct from execution ownership; all data are synthetic.""" import pytest import json +from copy import deepcopy from pathlib import Path from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, empty_todo_display @@ -58,6 +59,75 @@ def test_agent_execution_still_respects_claim_and_exclusion(): assert result["claim_scope"]["other_agent_claimed_open_count"] == 1 +@pytest.mark.parametrize("identity_field", ["target_key", "action_kind", "title"]) +def test_quota_frontier_keeps_identity_without_a_typed_todo_id(identity_field): + row = item(None, task_class="advancement_task", resume_when="resume_at:2099-01-01T00:00:00Z", + resume_ready=False, next_due_at="2099-01-01T00:00:00Z", + **{identity_field: "synthetic-frontier-identity"}) + result = summarize_user_todos_for_quota(summary([row]), + agent_identity={"agent_id": "agent-b"}) + assert result["frontier_deadline"]["identity"] == "synthetic-frontier-identity" + + +@pytest.mark.parametrize("entrypoint", ["summary", "planning"]) +def test_quota_returned_records_do_not_alias_source_or_other_lanes(entrypoint): + from loopx.control_plane.todos.quota_selection import project_quota_planning + + row = item("todo_wait", task_class="advancement_task", required_capabilities=["shell"], + resume_when="todo_done:todo_dependency", resume_ready=False, + resume_condition={"kind": "todo_done", "satisfied": False, + "target_todo_id": "todo_dependency", "target_status": "open", + "target_task_class": "advancement_task"}) + source = summary([row]) + before = deepcopy(source) + if entrypoint == "summary": + result = summarize_user_todos_for_quota(source, agent_identity={"agent_id": "agent-b"}) + output = result["resume_blocked_items"][0] + other = result["first_open_items"][0] + else: + result = project_quota_planning(source, all_open_items=[row], source_open_count=1, + agent_identity={"agent_id": "agent-b"}, filter_user_gate_blocks_agent=False, + available_capabilities=[], current_time="2026-10-01T00:00:00Z") + resume = result["resume_planning"] + output = resume["resume_blocked_lanes"]["resume_blocked_items"][0] + other = resume["blocked_successor_items"][0] + output["required_capabilities"].append("synthetic-change") + output["resume_condition"]["target_status"] = "done" + assert source == before + assert other["required_capabilities"] == ["shell"] + assert other["resume_condition"]["target_status"] == "open" + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_large_history_completes_public_quota_through_vision_wait(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md" + records = [item(f"todo_history_{index}", role="agent", source_section="Agent Todo", + schema_version="todo_item_v0", archive_state="active", done=True, status="done", + task_class="advancement_task", text=f"Completed synthetic work {index}. " * 60, + index=index + 1) for index in range(2500)] + records.append(item("todo_tail", role="agent", source_section="Agent Todo", + schema_version="todo_item_v0", archive_state="active", done=False, + task_class="advancement_task", claimed_by="agent-b", index=2501)) + state.write_text("---\nstatus: active\n---\n# Goal\n\n## Objective\n" + "Deliver a checked change.\n\n## Next Action\nComplete the remaining work.\n") + write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry, + goal_id="goal-history", domain="quota-history", adapter_kind="generic_project_goal_v0", + state_file=str(state), registered_agents=["agent-b"], quota_allowed_slots=None) + projection = build_todo_runtime_shadow_projection( + goal_id="goal-history", todos=records, handoff_mode="soft_claim") + initialize_canonical_authority(runtime, "goal-history", projection, state_path=state, provider=provider) + before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-history") + narrative = state.read_bytes() + code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "goal-history", + "--agent-id", "agent-b", registry_path=registry, runtime_root=runtime) + assert code == 0, packet + assert packet["agent_todo_summary"]["total_count"] == 2501 + assert packet["selected_todo"]["todo_id"] == "todo_tail" + assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-history") == before + assert state.read_bytes() == narrative + + @pytest.mark.parametrize("gate_scope", [False, True]) def test_large_quota_history_uses_snapshot_without_losing_scope( monkeypatch, gate_scope, diff --git a/tests/control_plane/test_resume_planning_projection.py b/tests/control_plane/test_resume_planning_projection.py index 3a3e585e67..75cd0658ea 100644 --- a/tests/control_plane/test_resume_planning_projection.py +++ b/tests/control_plane/test_resume_planning_projection.py @@ -13,8 +13,10 @@ from loopx.control_plane.todos.resume_planning import ( build_referenced_todo_resume_planning_request, + build_todo_resume_planning_request, project_todo_resume_planning, ) +from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result from loopx.control_plane.todos.quota_summary import select_quota_todo_summary @@ -56,12 +58,12 @@ def ids(items: list[dict]) -> list[str]: return [item["todo_id"] for item in items] -def test_referenced_request_deduplicates_lanes_and_keeps_full_payload_local() -> None: +def test_referenced_request_deduplicates_lanes_without_pruning_payload_facts() -> None: row = waiting( "todo_shared", content_revision="sha256:synthetic-full-projection", ) - request, payloads = build_referenced_todo_resume_planning_request({ + request = build_referenced_todo_resume_planning_request({ "items": [row], "first_open_items": [row], "deferred_items": [row], @@ -72,9 +74,34 @@ def test_referenced_request_deduplicates_lanes_and_keeps_full_payload_local() -> assert request["sources"]["items"] == [0] assert request["sources"]["first_open_items"] == [0] assert request["sources"]["deferred_items"] == [0] - assert request["records"][0]["payload"]["_loopx_resume_record_ref"] == 0 - assert "content_revision" not in request["records"][0]["payload"] - assert payloads[0]["content_revision"] == row["content_revision"] + assert request["records"][0]["payload"]["content_revision"] == row["content_revision"] + + +def test_referenced_projection_keeps_conflicting_same_id_source_facts() -> None: + source = {"items": [waiting("todo_shared")], + "first_open_items": [waiting("todo_shared", claimed_by="agent-b")], + "deferred_items": [waiting("todo_shared", status="deferred")]} + request = build_referenced_todo_resume_planning_request(source, agent_id="agent-a") + assert len(request["records"]) == 3 + assert effect_runtime_result("todo.resume_planning.project", request) == effect_runtime_result( + "todo.resume_planning.project", build_todo_resume_planning_request(source, agent_id="agent-a")) + + +@pytest.mark.parametrize("reference", [-1, 1, True, 0.5, "0"]) +def test_referenced_projection_rejects_malformed_or_out_of_range_references(reference) -> None: + request = build_referenced_todo_resume_planning_request({"items": [waiting("todo_shared")]}) + request["sources"]["items"] = [reference] + with pytest.raises(EffectRuntimeRejected, match="record reference"): + effect_runtime_result("todo.resume_planning.project", request) + + +@pytest.mark.parametrize("schema", [["todo_resume_planning_request_v0"], + ["todo_resume_planning_request_v1"], None, "unsupported"]) +def test_resume_planning_rejects_coercible_or_unknown_schema_versions(schema) -> None: + request = build_referenced_todo_resume_planning_request({}) + request["schema_version"] = schema + with pytest.raises(EffectRuntimeRejected, match="schema mismatch"): + effect_runtime_result("todo.resume_planning.project", request) def test_claim_priority_and_exclusion_are_independent_of_display_priority() -> None: diff --git a/tests/control_plane/test_vision_wait_coverage.py b/tests/control_plane/test_vision_wait_coverage.py index a506540d55..53d8d71380 100644 --- a/tests/control_plane/test_vision_wait_coverage.py +++ b/tests/control_plane/test_vision_wait_coverage.py @@ -105,9 +105,9 @@ def test_vision_wait_composition_uses_one_runtime_crossing(monkeypatch) -> None: methods = [] original = goal_vision_wait.effect_runtime_result - def counted(method, request): + def counted(method, request, **kwargs): methods.append(method) - return original(method, request) + return original(method, request, **kwargs) monkeypatch.setattr(goal_vision_wait, "effect_runtime_result", counted) decision() From eda29472f2cc69ca99161651d463415491f8ea20 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 29 Oct 2026 15:11:56 +0800 Subject: [PATCH 3/3] docs(architecture): qualify complete large-goal quota transport Signed-off-by: huangruiteng --- .../rfcs/typescript-control-plane-migration-v0.md | 10 ++++++++++ .../typescript-control-plane-migration-v0.zh-CN.md | 8 ++++++++ 2 files changed, 18 insertions(+) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 2d9ac52f66..7e4643039f 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1450,6 +1450,16 @@ unknown-effect policy are unchanged; decoded expansion uses the existing 64 MiB local-snapshot bound before copying records. This is internal transport, not a new capability, setting, authority source or installed-runtime claim. +The resume-planning input uses a co-deployed v1 record table for repeated source +lanes. It retains every v0 payload fact before typed decisions, rather than a +consumer-specific field allowlist or a second Python restoration pass. Existing +v0 callers remain valid; conflicting same-id facts are not merged. Both quota +planning and its subsequent Vision-wait projection opt into the existing private, +digest-verified local snapshot transport for unique histories beyond 2 MiB; +the 64 MiB snapshot/expanded-response bounds and ordinary RPC limits remain. +File/SQLite qualification must reach the public quota result, preserve the source +and select tail work, not stop after the first oversized request is repaired. + Current bounded delivery: shared-goal alignment and amendment admission use one `shared_goal_work_source.py` snapshot per decision, reusing the canonical Todo summary after promotion. The same provider read optionally supplies leases at 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 93799409a8..fe74828ebd 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 @@ -1045,6 +1045,14 @@ lane 正文超过 2 MiB 响应边界、从而阻断启动及 Goal 增长后的 64 MiB 展开边界。这是内部传输，不新增 capability、设置、授权源，也不代表安装态 已采用此修改。 +Resume-planning 输入采用协同发布的 v1 record table，供重复 source lane 引用。 +每条记录在 TS 决策前保留全部 v0 payload 事实，不使用面向某个消费者的字段白名单， +也不再由 Python 二次恢复。已有 v0 调用者仍有效；相同 id 的冲突事实不会合并。 +Quota planning 及其后续 Vision-wait 投影均显式复用既有私有、摘要校验的本地 +snapshot 通道，承载超过 2 MiB 的独立历史；64 MiB snapshot／响应展开边界和 +普通 RPC 上限不变。File／SQLite 验证必须到达公开 quota 结果，保留来源并选中 +尾部工作，不能只以第一层超大请求得到修复作为完整交付证据。 + 当前有边界交付：shared-goal alignment 与 amendment admission 每次决策共用一份 `shared_goal_work_source.py` 快照，promotion 后复用 canonical Todo summary；同一次 provider 读取可返回同 revision 的 lease。缺失／空／陈旧展示及旧 lease 文件不再是