From da21d1f1d63a80ca5060beb752c7c97b450c65ca Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 02:00:23 +0800 Subject: [PATCH 1/3] refactor(todos): own complete summary decisions in TypeScript Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../todo-projection-shared-helper-smoke.py | 18 +- .../control_plane/effect_runtime_handlers.ts | 8 +- .../control_plane/todos/summary_projection.ts | 152 ++++++++ loopx/control_plane/todos/todo_semantics.py | 50 --- loopx/control_plane/todos/todo_summary.py | 360 ++++-------------- .../project_registry_io_manifest_v1.json | 8 +- loopx/status.py | 12 +- .../control_plane/test_todo_consumer_scope.py | 10 +- .../test_todo_summary_projection.py | 62 +++ .../todo_consumer_scope_conformance.ts | 9 + .../todo_summary_projection.test.ts | 102 +++++ 11 files changed, 427 insertions(+), 364 deletions(-) create mode 100644 loopx/control_plane/todos/summary_projection.ts create mode 100644 tests/control_plane/test_todo_summary_projection.py create mode 100644 tests/control_plane_ts/todo_summary_projection.test.ts diff --git a/examples/control_plane/todo-projection-shared-helper-smoke.py b/examples/control_plane/todo-projection-shared-helper-smoke.py index 1f70c34dce..e77b11c1b3 100644 --- a/examples/control_plane/todo-projection-shared-helper-smoke.py +++ b/examples/control_plane/todo-projection-shared-helper-smoke.py @@ -33,7 +33,6 @@ build_goal_frontier_projection_from_summaries, ) from loopx.status import ( # noqa: E402 - claimed_visibility_items as status_claimed_visibility_items, todo_item_is_deferred as status_todo_item_is_deferred, todo_projection_sort_key, ) @@ -42,7 +41,6 @@ TODO_TASK_CLASS_MONITOR, ) from loopx.control_plane.todos.projection import ( # noqa: E402 - todo_claimed_visibility_items as shared_claimed_visibility_items, todo_item_claimed_by_agent_or_unclaimed as shared_todo_item_claimed_by_agent_or_unclaimed, todo_item_is_deferred as shared_todo_item_is_deferred, todo_projection_sort_key as shared_todo_projection_sort_key, @@ -165,7 +163,7 @@ def assert_shared_ordering_parity(summary: dict) -> None: assert quota_todo_projection_sort_key(embedded_priority) == (50, 9), embedded_priority -def assert_claimed_visibility_parity() -> None: +def assert_claim_scope_parity() -> None: items = [ quota_todo_item( todo_id="todo_a1", @@ -199,18 +197,6 @@ def assert_claimed_visibility_parity() -> None: task_class=TODO_TASK_CLASS_ADVANCEMENT, ), ] - for selector in ( - shared_claimed_visibility_items, - status_claimed_visibility_items, - ): - selected_two = selector(items, limit=2) - assert [item["todo_id"] for item in selected_two] == ["todo_a1", "todo_b1"], selected_two - selected_three = selector(items, limit=3) - assert [item["todo_id"] for item in selected_three] == [ - "todo_a1", - "todo_a2", - "todo_b1", - ], selected_three claimed_by_current = items[0] claimed_by_other = items[2] unclaimed = items[3] @@ -487,7 +473,7 @@ def main() -> int: summary = build_agent_todo_summary() assert_status_summary_lanes(summary) assert_shared_ordering_parity(summary) - assert_claimed_visibility_parity() + assert_claim_scope_parity() assert_agent_scope_frontier_routing_parity() assert_deferred_helper_parity() assert_monitor_item_collection_parity(summary) diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 558bcdb980..df22465742 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -1,3 +1,4 @@ +import {projectTodoSummary} from "./todos/summary_projection.ts"; import {readShadowDrainPlan} from "./coordination/shadow_drain_plan.ts"; import {manageAutomationCadence, projectCadenceSchedule} from "./quota/automation_cadence.ts"; import {manageLocalAuthorityArchive} from "./coordination/local_authority_archive.ts"; @@ -7,8 +8,8 @@ import {projectPeerOrchestration} from "./quota/peer_orchestration.ts"; import {inspectTaskLease} from "./work_items/task_lease_inspection.ts"; import {evaluateTodoPriority} from "./todos/priority.ts"; import {evaluateUserCompletion} from "./todos/user_completion.ts"; -import {projectTodoSuccession, projectTodoClosure} from "./todos/succession.ts"; -import {projectTodoSummaryLanes, projectLegacyTodoWorkCounts} from "./todos/summary_lanes.ts"; +import {projectTodoSuccession} from "./todos/succession.ts"; +import {projectLegacyTodoWorkCounts} from "./todos/summary_lanes.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"; @@ -435,11 +436,10 @@ export function createEffectRuntimeHandlers( ["todo.priority.plan", evaluateTodoPriority], ["todo.public_update.plan", planPublicTodoUpdate], ["todo.standing_decision.project", evaluateStandingDecisionProjection], - ["todo.summary_lanes.project", projectTodoSummaryLanes], + ["todo.summary.project", projectTodoSummary], ["capabilities.periodic_report.progress.select", selectPeriodicReportProgress], ["capabilities.periodic_report.approval_retry.select", selectPeriodicReportApprovalRetry], ["todo.succession.project", projectTodoSuccession], - ["todo.succession.closure", projectTodoClosure], ["todo.work_counts.project", projectLegacyTodoWorkCounts], ["todo.decision_scope.evaluate", evaluateDecisionScope], ["todo.user_completion.plan", evaluateUserCompletion], diff --git a/loopx/control_plane/todos/summary_projection.ts b/loopx/control_plane/todos/summary_projection.ts new file mode 100644 index 0000000000..7dbf0a7e36 --- /dev/null +++ b/loopx/control_plane/todos/summary_projection.ts @@ -0,0 +1,152 @@ +/** Whole-source summary decisions. Python materializes these ordinals as public + * display records; counts and closure never depend on a presentation budget. */ +import type {JsonObject} from "../effect_program.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; +import {requireBoolean, requireJsonObject, requireStringLiteral} from "../runtime_decode.ts"; +import {parseTodoTimestampMicros} from "../runtime_timestamp.ts"; +import {projectTodoSummaryLanes, type TodoSummaryLane} from "./summary_lanes.ts"; +import {projectTodoClosure} from "./succession.ts"; + +type Format = "raw" | "compact" | "active" | "recent" | "gap"; +interface DisplayLane {indices: number[]; format: Format} +interface SummaryProjection { + schema_version: "todo_summary_projection_v0"; + source_indices: number[]; + full_selection: boolean; + fields: JsonObject; + lanes: Record; + orchestration: {candidate_items: number[]; user_blocker_items: number[]}; +} + +/** Allocate a bounded display across claimants, then restore source ordering. */ +function claimedVisibility(indices: readonly number[], rows: readonly JsonObject[], limit: number): number[] { + if (indices.length <= limit) return [...indices]; + const buckets = new Map(); + for (const index of indices) { + const claim = rows[index].claim; + if (typeof claim !== "string" || !claim) continue; + const bucket = buckets.get(claim) ?? []; + bucket.push(index); buckets.set(claim, bucket); + } + if (!buckets.size) return indices.slice(0, limit); + const perClaimant = Math.max(1, Math.floor(limit / buckets.size)); + const selected = new Set(); + for (const bucket of buckets.values()) { + for (const index of bucket.slice(0, perClaimant)) { + if (selected.size < limit) selected.add(index); + } + } + for (const index of indices) if (selected.size < limit) selected.add(index); + return indices.filter(index => selected.has(index)); +} + +export function projectTodoSummary(value: unknown): SummaryProjection { + const request = requireJsonObject(value, "Todo summary request"); + if (request.schema_version !== "todo_summary_projection_request_v0" || !Array.isArray(request.rows)) { + throw new EffectRuntimeRequestError("Todo summary request schema mismatch"); + } + const role = request.role === null ? null : requireStringLiteral(request.role, ["user", "agent"], "role"); + if (request.source_section !== null && typeof request.source_section !== "string") { + throw new EffectRuntimeRequestError("source_section must be text or null"); + } + const limit = request.item_limit; + if (limit !== null && (typeof limit !== "number" || !Number.isSafeInteger(limit) || limit < 0)) { + throw new EffectRuntimeRequestError("item_limit must be a non-negative integer or null"); + } + const full = requireBoolean(request.full_selection, "full_selection"); + const rows = request.rows.map(value => requireJsonObject(value, "summary row")); + // The co-deployed adapter sends source facts, not prose or full Todo bodies. + for (const row of rows) { + if ((row.claim !== null && typeof row.claim !== "string") || row.claimed !== Boolean(row.claim)) { + throw new EffectRuntimeRequestError("summary claimant facts disagree"); + } + requireBoolean(row.linked_user_action, "linked_user_action"); + if ((row.completed_at !== null && typeof row.completed_at !== "string") || + (row.updated_at !== null && typeof row.updated_at !== "string") || + typeof row.completion_index !== "number" || !Number.isSafeInteger(row.completion_index)) { + throw new EffectRuntimeRequestError("invalid summary completion coordinates"); + } + } + const hasSelection = request.selection !== null; + const projected = projectTodoSummaryLanes({ + schema_version: hasSelection ? "todo_summary_lanes_request_v1" : "todo_summary_lanes_request_v0", + rows, observed_at: request.observed_at, ...(hasSelection ? {selection: request.selection} : {}), + }); + const source = hasSelection ? projected.source_indices as number[] : rows.map((_, index) => index); + const fullSelection = full && (!hasSelection || projected.full_selection === true); + const selected = projected.lanes as Record; + const fields: JsonObject = {schema_version: "todo_summary_v0", source_section: request.source_section, + total_count: source.length, work_counts: projected.work_counts, + open_count: selected.open_items.length, done_count: selected.terminal_items.length, + advancement_done_count: selected.done_items.filter(index => rows[index].task_class === "advancement_task").length, + deferred_count: selected.deferred_items.length, monitor_due_count: selected.monitor_due_items.length, + monitor_schedule_gap_count: selected.monitor_schedule_gap_items.length, + }; + const lanes: Record = {}; + const lane = (name: string, indices: readonly number[], cap: number | null = null, format: Format = "compact") => { + lanes[name] = {indices: cap === null ? [...indices] : indices.slice(0, cap), format}; + }; + lane("first_open_items", selected.projected_open_items, 3); + lane("first_executable_items", selected.executable_items, 3); + lane("monitor_open_items", selected.monitor_items); + lane("monitor_due_items", selected.monitor_due_items, 1); + lane("monitor_schedule_gap_items", selected.monitor_schedule_gap_items, 1); + lane("unclaimed_priority_open_items", selected.unclaimed_open_items, 8); + lane("claimed_open_items", claimedVisibility(selected.claimed_open_items, rows, 16)); + lane("claimed_advancement_open_items", claimedVisibility(selected.claimed_advancement_items, rows, 16)); + lane("claimed_monitor_open_items", claimedVisibility(selected.claimed_monitor_items, rows, 16)); + lane("backlog_items", selected.projected_open_items, 8); + lane("executable_backlog_items", selected.executable_items, 8); + lane("deferred_items", selected.projected_deferred_items, 8); + lane("deferred_resume_candidates", selected.projected_deferred_items.filter(index => rows[index].resume_ready === true), 8); + lane("items", selected.budgeted_items, limit, "raw"); + + // Recent completion is chronological, not ISO-string order or last-edit order. + // Unknown legacy times remain in totals, but cannot claim a recent timestamp. + const instant = (value: unknown): bigint | null => typeof value === "string" && value.trim() + ? parseTodoTimestampMicros(value.trim()) : null; + const completedAt = rows.map(row => instant(row.completed_at)); + const changedAt = rows.map(row => instant(row.updated_at || row.completed_at)); + const timestamp = (index: number, completionOnly: boolean) => (completionOnly ? completedAt : changedAt)[index]; + const byTime = (completionOnly: boolean) => (left: number, right: number) => { + const a = timestamp(left, completionOnly), b = timestamp(right, completionOnly); + if (a !== b) return a === null ? 1 : b === null ? -1 : a > b ? -1 : 1; + return Number(rows[right].completion_index) - Number(rows[left].completion_index); + }; + const recent = selected.done_items.filter(index => rows[index].task_class === "advancement_task" && timestamp(index, true) !== null) + .sort(byTime(true)); + if (recent.length) lane("recent_completed_advancement_items", recent, 16, "recent"); + const gaps = source.filter(index => rows[index].successor_gap === true).sort(byTime(false)); + if (gaps.length) { + fields.completed_without_successor_count = gaps.length; + lane("completed_without_successor_items", gaps, 5, "gap"); + } + if (selected.watch_only_monitor_items.length) { + fields.watch_only_monitor_count = selected.watch_only_monitor_items.length; + fields.watch_only_monitor_due_count = selected.watch_only_monitor_due_items.length; + fields.convergence_open_count = selected.convergent_open_items.length; + } + for (const [name, indices, cap] of [ + ["blocker", selected.blocker_items, null], ["resume_blocked", selected.resume_blocked_items, 8], + ] as const) { + if (indices.length) { + fields[name === "blocker" ? "blocker_open_count" : "resume_blocked_count"] = indices.length; + lane(`${name}_items`, indices, cap); + } + } + if (selected.active_next_action_items.length) lane("active_next_action_items", selected.active_next_action_items, null, "active"); + if (selected.active_next_action_executable_items.length) lane("active_next_action_executable_items", selected.active_next_action_executable_items, null, "active"); + if (selected.claimed_open_items.length) { + fields.claimed_open_count = selected.claimed_open_items.length; + fields.unclaimed_open_count = selected.open_items.length - selected.claimed_open_items.length; + fields.claimed_advancement_open_count = selected.claimed_advancement_items.length; + fields.claimed_monitor_open_count = selected.claimed_monitor_items.length; + } + Object.assign(fields, projectTodoClosure({schema_version: "todo_closure_request_v0", role, + source_section: request.source_section, full_selection: fullSelection, rows: source.map(index => rows[index])})); + return {schema_version: "todo_summary_projection_v0", source_indices: source, full_selection: fullSelection, fields, lanes, + orchestration: { + candidate_items: role === "agent" ? selected.projected_open_items.filter(index => rows[index].task_class === "advancement_task") : [], + user_blocker_items: role === "user" ? selected.projected_open_items.filter(index => rows[index].linked_user_action === true) : [], + }}; +} diff --git a/loopx/control_plane/todos/todo_semantics.py b/loopx/control_plane/todos/todo_semantics.py index 46315418d5..4b34911da0 100644 --- a/loopx/control_plane/todos/todo_semantics.py +++ b/loopx/control_plane/todos/todo_semantics.py @@ -178,56 +178,6 @@ def todo_projection_sort_key( return (todo_priority_rank(item, text_mode=text_mode), todo_index_rank(item)) -def todo_claimed_visibility_items( - items: list[dict[str, Any]], - *, - limit: int, -) -> list[dict[str, Any]]: - if limit <= 0 or len(items) <= limit: - return items[:limit] - claim_order: list[str] = [] - buckets: dict[str, list[dict[str, Any]]] = {} - for item in items: - claimed_by = normalize_todo_claimed_by(item.get("claimed_by")) - if not claimed_by: - continue - if claimed_by not in buckets: - buckets[claimed_by] = [] - claim_order.append(claimed_by) - buckets[claimed_by].append(item) - if not buckets: - return items[:limit] - - original_index = {id(item): index for index, item in enumerate(items)} - per_claimant_cap = max(1, limit // len(buckets)) - selected: list[dict[str, Any]] = [] - selected_ids: set[int] = set() - for claimed_by in claim_order: - taken = 0 - for item in buckets[claimed_by]: - if taken >= per_claimant_cap: - break - if len(selected) >= limit: - break - selected.append(item) - selected_ids.add(id(item)) - taken += 1 - if len(selected) >= limit: - break - - if len(selected) < limit: - for item in items: - if id(item) in selected_ids: - continue - selected.append(item) - selected_ids.add(id(item)) - if len(selected) >= limit: - break - - return sorted( - selected, key=lambda item: original_index.get(id(item), TODO_MISSING_INDEX) - )[:limit] - def todo_item_task_text( item: dict[str, Any], diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py index 6b74353c78..9f6cbc9763 100644 --- a/loopx/control_plane/todos/todo_summary.py +++ b/loopx/control_plane/todos/todo_summary.py @@ -1,6 +1,5 @@ from __future__ import annotations -from dataclasses import dataclass from datetime import datetime import re from typing import Any, Callable, Optional @@ -45,7 +44,6 @@ from .handoff_gate import build_todo_handoff_gate_states from .handoff_note import attach_todo_handoff_note from .todo_semantics import ( - todo_claimed_visibility_items as projection_todo_claimed_visibility_items, todo_item_is_actionable_open as projection_todo_item_is_actionable_open, todo_item_is_deferred as projection_todo_item_is_deferred, todo_item_is_due_monitor as projection_todo_item_is_due_monitor, @@ -84,7 +82,6 @@ MAX_MONITOR_DUE_ITEMS = 1 MAX_DEPENDENCY_BLOCKERS = 4 MAX_COMPLETED_SUCCESSION_WARNING_ITEMS = 5 -MAX_RECENT_COMPLETED_ADVANCEMENT_ITEMS = MAX_TODO_VISIBILITY_LANE_ITEMS TASK_ORCHESTRATION_AUTHORITY_SCHEMA_VERSION = "task_orchestration_authority_v0" TODO_ARCHIVE_STATE_ACTIVE = "active" @@ -95,31 +92,6 @@ FirstOpenTodoText = Callable[[Optional[dict[str, Any]]], Optional[str]] -@dataclass(frozen=True) -class _TodoGroupLanes: - open_items: list[dict[str, Any]] - terminal_items: list[dict[str, Any]] - deferred_items: list[dict[str, Any]] - done_items: list[dict[str, Any]] - projected_open_items: list[dict[str, Any]] - projected_deferred_items: list[dict[str, Any]] - budgeted_items: list[dict[str, Any]] - claimed_open_items: list[dict[str, Any]] - unclaimed_open_items: list[dict[str, Any]] - executable_items: list[dict[str, Any]] - blocker_items: list[dict[str, Any]] - resume_blocked_items: list[dict[str, Any]] - monitor_items: list[dict[str, Any]] - monitor_due_items: list[dict[str, Any]] - watch_only_monitor_items: list[dict[str, Any]] - watch_only_monitor_due_items: list[dict[str, Any]] - non_watch_only_monitor_due_items: list[dict[str, Any]] - convergent_open_items: list[dict[str, Any]] - monitor_schedule_gap_items: list[dict[str, Any]] - claimed_advancement_items: list[dict[str, Any]] - claimed_monitor_items: list[dict[str, Any]] - active_next_action_items: list[dict[str, Any]] - active_next_action_executable_items: list[dict[str, Any]] TASK_ORCHESTRATION_CANDIDATE_FIELDS = ( "todo_id", "status", @@ -494,45 +466,13 @@ def canonical_todo_read_record( return record -def _task_orchestration_authority( - lanes: _TodoGroupLanes, - *, - role: str | None, -) -> dict[str, Any]: - candidate_items = ( - [ - { - key: compact[key] - for key in TASK_ORCHESTRATION_CANDIDATE_FIELDS - if key in compact - } - for item in lanes.projected_open_items - if todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT - for compact in [compact_todo_item(item)] - ] - if role == "agent" - else [] - ) - user_blocker_items = ( - [ - { - key: compact[key] - for key in TASK_ORCHESTRATION_USER_BLOCKER_FIELDS - if key in compact - } - for item in lanes.projected_open_items - if normalize_todo_id(item.get("unblocks_todo_id")) - for compact in [compact_todo_item(item)] - ] - if role == "user" - else [] - ) - return { - "schema_version": TASK_ORCHESTRATION_AUTHORITY_SCHEMA_VERSION, - "role": role, - "candidate_items": candidate_items, - "user_blocker_items": user_blocker_items, - } +def _task_orchestration_authority(lanes: dict[str, list[dict[str, Any]]], *, role: str | None) -> dict[str, Any]: + """Materialize the typed selection using the existing public field allowlist.""" + return {"schema_version": TASK_ORCHESTRATION_AUTHORITY_SCHEMA_VERSION, "role": role, + **{name: [{key: compact[key] for key in fields if key in compact} + for item in lanes[name] for compact in [compact_todo_item(item)]] + for name, fields in (("candidate_items", TASK_ORCHESTRATION_CANDIDATE_FIELDS), + ("user_blocker_items", TASK_ORCHESTRATION_USER_BLOCKER_FIELDS))}} def compact_active_next_action_todo_item(item: dict[str, Any]) -> dict[str, Any]: @@ -591,10 +531,6 @@ def todo_projection_sort_key(item: dict[str, Any]) -> tuple[int, int]: return projection_todo_projection_sort_key(item, text_mode="prefix") -def claimed_visibility_items(items: list[dict[str, Any]], *, limit: int) -> list[dict[str, Any]]: - return projection_todo_claimed_visibility_items(items, limit=limit) - - def todo_item_is_deferred(item: dict[str, Any]) -> bool: return projection_todo_item_is_deferred(item) @@ -825,32 +761,6 @@ def todo_item_is_succession_tracked_completion(item: dict[str, Any]) -> bool: return project_succession([item])[0]["tracked_completion"] is True -def _completed_succession_sort_key(item: dict[str, Any]) -> tuple[str, int]: - raw_index = item.get("index") - try: - index = int(raw_index) if raw_index is not None else 0 - except (TypeError, ValueError): - index = 0 - timestamp = str(item.get("updated_at") or item.get("completed_at") or "") - return (timestamp, index) - - -def completed_without_successor_items( - items: list[dict[str, Any]], *, evaluations: list[dict[str, Any]], -) -> list[dict[str, Any]]: - gap_items = [] - for item, evaluation in zip(items, evaluations, strict=True): - if not evaluation["successor_gap"]: - continue - compact = compact_todo_item(item) - for key in ("note", "evidence", "reason"): - compact.pop(key, None) - compact["succession_tracked"] = True - compact["recommended_action"] = "record no_followup=true or add/link a successor todo" - gap_items.append(compact) - return sorted(gap_items, key=_completed_succession_sort_key, reverse=True) - - def _structured_todo_group_items( items: list[dict[str, Any]], *, @@ -895,14 +805,21 @@ def _structured_resume_source_items( ] -def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None, - selection: dict[str, Any] | None = None, +def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None, + *, selection: dict[str, Any] | None, role: str | None, source_section: str | None, + item_limit: int | None, full_selection: bool, ) -> dict[str, Any]: - """Adapt legacy facts and read batch ordinals from the typed lane owner.""" + """Adapt evaluated facts and materialize one typed summary decision.""" from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result + from .succession_warning import project_succession + + succession = project_succession(items, reuse=True) + handoff_gates = build_todo_handoff_gate_states(items, evaluations=succession) + replan_gates = {gate.get("todo_id") for gate in handoff_gates + if gate.get("route_continuation_replan_required") is True} rows = [] - for item in items: + for item, evaluation in zip(items, succession, strict=True): resume = normalize_todo_resume_when(item.get("resume_when")) condition = item.get("resume_condition") evaluated = (isinstance(condition, dict) @@ -921,41 +838,78 @@ def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[ "watch_only": projection_todo_item_is_watch_only_monitor(item), "due_at": due.timestamp() if due else None, "expires_at": expires.timestamp() if expires else None, "sort": list(projection_todo_presentation_sort_key(item)), - **({"todo_id": normalize_todo_id(item.get("todo_id")), + "completed_at": str(item.get("completed_at") or "") or None, + "updated_at": str(item.get("updated_at") or "") or None, + "completion_index": int(item.get("index") or 0), + "linked_user_action": bool(normalize_todo_id(item.get("unblocks_todo_id"))), + "no_followup": normalize_todo_no_followup(item.get("no_followup")) is True, + "successor_gap": evaluation["successor_gap"], "handoff_state": evaluation["handoff_state"], + "replan": item.get("route_continuation_replan_required") is True or item.get("todo_id") in replan_gates, + **{"todo_id": normalize_todo_id(item.get("todo_id")), "claim": normalize_todo_claimed_by(item.get("claimed_by")), "bound": normalize_todo_bound_agent(item.get("bound_agent")), "blocks": normalize_todo_blocks_agent(item.get("blocks_agent")), "global": bool(item.get("global_gate")), - "excluded": normalize_todo_excluded_agents(item.get("excluded_agents"))} - if selection is not None else {})}) + "excluded": normalize_todo_excluded_agents(item.get("excluded_agents"))}}) try: - result = effect_runtime_result("todo.summary_lanes.project", { - "schema_version": "todo_summary_lanes_request_v0" if selection is None else "todo_summary_lanes_request_v1", + result = effect_runtime_result("todo.summary.project", { + "schema_version": "todo_summary_projection_request_v0", "rows": rows, "observed_at": now_utc().timestamp(), - **({"selection": selection} if selection is not None else {}), + "selection": selection, "role": role, "source_section": source_section, + "item_limit": item_limit, "full_selection": full_selection, }) except EffectRuntimeRejected as error: raise ValueError(str(error)) from error - if not isinstance(result, dict) or result.get("schema_version") != "todo_summary_lanes_v0": - raise ValueError("invalid typed Todo summary lanes") + if not isinstance(result, dict) or result.get("schema_version") != "todo_summary_projection_v0": + raise ValueError("invalid typed Todo summary projection") + def valid_ordinals(value: Any) -> bool: return (isinstance(value, list) and all(type(index) is int and 0 <= index < len(items) for index in value) and len(set(value)) == len(value)) - lanes = result["lanes"] - if not isinstance(lanes, dict) or any(not valid_ordinals(indices) for indices in lanes.values()): - raise ValueError("invalid Todo summary source ordinal") - selected = result.get("source_indices", list(range(len(items)))) - if (not valid_ordinals(selected) - or selection is not None and ("source_indices" not in result or type(result.get("full_selection")) is not bool)): + selected = result.get("source_indices") + if not valid_ordinals(selected) or type(result.get("full_selection")) is not bool: raise ValueError("invalid typed Todo selection ordinals") selected_set = set(selected) - if any(not set(indices) <= selected_set for indices in lanes.values()): - raise ValueError("Todo summary lane escaped the selected source") - return {"lanes": {key: [items[index] for index in indices] for key, indices in lanes.items()}, - "items": [items[index] for index in selected], - "full_selection": result.get("full_selection", True), "work_counts": result["work_counts"]} + lanes, orchestration = result.get("lanes"), result.get("orchestration") + if not isinstance(lanes, dict) or not isinstance(orchestration, dict): + raise ValueError("invalid typed Todo summary lanes") + for indices in [*(lane.get("indices") if isinstance(lane, dict) else None for lane in lanes.values()), + *orchestration.values()]: + if not valid_ordinals(indices): + raise ValueError("invalid Todo summary source ordinal") + if not set(indices) <= selected_set: + raise ValueError("Todo summary lane escaped the selected source") + summary = result.get("fields") + if not isinstance(summary, dict) or summary.get("schema_version") != "todo_summary_v0": + raise ValueError("invalid typed Todo summary fields") + for name, lane in lanes.items(): + mode = lane.get("format") + if mode not in {"raw", "active", "compact", "recent", "gap"}: + raise ValueError("invalid Todo summary display format") + formatted = [] + for index in lane["indices"]: + item = items[index] + if mode == "raw": + compact = item + elif mode == "active": + compact = compact_active_next_action_todo_item(item) + elif mode in {"compact", "recent", "gap"}: + compact = compact_todo_item(item) + if mode in {"recent", "gap"}: + for key in ("note", "evidence", "reason"): + compact.pop(key, None) + if mode == "gap": + compact.update(succession_tracked=True, + recommended_action="record no_followup=true or add/link a successor todo") + else: + raise ValueError("invalid Todo summary display format") + formatted.append(compact) + summary[name] = formatted + return {"summary": summary, "items": [items[index] for index in selected], + "succession": [succession[index] for index in selected], + "orchestration": {name: [items[index] for index in indices] for name, indices in orchestration.items()}} def compact_todo_group( @@ -1023,183 +977,33 @@ def compact_evaluated_todo_group( """ if not items and not include_empty_source: return None - projected = _project_summary_lanes(items, preferred_todo_ids, selection) + projected = _project_summary(items, preferred_todo_ids, selection=selection, + role=role, source_section=source_section, item_limit=item_limit, full_selection=full_selection) items = projected["items"] - if selection is not None: - full_selection = projected["full_selection"] if not items and not include_empty_source: return None - lanes = _TodoGroupLanes(**projected["lanes"]) - from .succession_warning import project_succession - - succession = project_succession(items, reuse=True) - successor_gap_items = completed_without_successor_items(items, evaluations=succession) - recent_completed_advancement_items = [ - compact_todo_item(item) - for item in sorted( - ( - item - for item in lanes.done_items - if todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT - and str(item.get("completed_at") or "").strip() - ), - key=_completed_succession_sort_key, - reverse=True, - )[:MAX_RECENT_COMPLETED_ADVANCEMENT_ITEMS] - ] - for item in recent_completed_advancement_items: - for key in ("note", "evidence", "reason"): - item.pop(key, None) - handoff_gates = build_todo_handoff_gate_states(items, evaluations=succession) - watch_only_monitor_items = lanes.watch_only_monitor_items - watch_only_monitor_due_items = lanes.watch_only_monitor_due_items - convergent_open_items = lanes.convergent_open_items - summary: dict[str, Any] = { - "schema_version": "todo_summary_v0", - "source_section": source_section, - "total_count": len(items), - "work_counts": projected["work_counts"], - "open_count": len(lanes.open_items), - "done_count": len(lanes.terminal_items), - "advancement_done_count": count_advancement_todos(lanes.done_items), - "deferred_count": len(lanes.deferred_items), - "first_open_items": [ - compact_todo_item(item) for item in lanes.projected_open_items[:3] - ], - "first_executable_items": [ - compact_todo_item(item) for item in lanes.executable_items[:3] - ], - "monitor_open_items": [ - compact_todo_item(item) for item in lanes.monitor_items - ], - "monitor_due_count": len(lanes.monitor_due_items), - "monitor_due_items": [ - compact_todo_item(item) - for item in lanes.monitor_due_items[:MAX_MONITOR_DUE_ITEMS] - ], - "monitor_schedule_gap_count": len(lanes.monitor_schedule_gap_items), - "monitor_schedule_gap_items": [ - compact_todo_item(item) - for item in lanes.monitor_schedule_gap_items[:MAX_MONITOR_DUE_ITEMS] - ], - "unclaimed_priority_open_items": [ - compact_todo_item(item) - for item in lanes.unclaimed_open_items[:MAX_PROJECT_ASSET_TODO_BACKLOG_ITEMS] - ], - "claimed_open_items": [ - compact_todo_item(item) - for item in claimed_visibility_items( - lanes.claimed_open_items, - limit=MAX_TODO_VISIBILITY_LANE_ITEMS, - ) - ], - "claimed_advancement_open_items": [ - compact_todo_item(item) - for item in claimed_visibility_items( - lanes.claimed_advancement_items, - limit=MAX_TODO_VISIBILITY_LANE_ITEMS, - ) - ], - "claimed_monitor_open_items": [ - compact_todo_item(item) - for item in claimed_visibility_items( - lanes.claimed_monitor_items, - limit=MAX_TODO_VISIBILITY_LANE_ITEMS, - ) - ], - "backlog_items": [ - compact_todo_item(item) - for item in lanes.projected_open_items[:MAX_PROJECT_ASSET_TODO_BACKLOG_ITEMS] - ], - "executable_backlog_items": [ - compact_todo_item(item) - for item in lanes.executable_items[:MAX_PROJECT_ASSET_TODO_BACKLOG_ITEMS] - ], - "deferred_items": [ - compact_todo_item(item) - for item in lanes.projected_deferred_items[:MAX_DEFERRED_TODO_VISIBILITY_ITEMS] - ], - "deferred_resume_candidates": [ - compact_todo_item(item) - for item in lanes.projected_deferred_items - if item.get("resume_ready") is True - ][:MAX_DEFERRED_TODO_VISIBILITY_ITEMS], - "items": lanes.budgeted_items if item_limit is None else lanes.budgeted_items[:item_limit], - } + summary = projected["summary"] + handoff_gates = build_todo_handoff_gate_states(items, evaluations=projected["succession"]) attach_advancement_frontier_revision_index(summary, items, role=role) attach_active_vision_waits( summary, vision_runs, role=role, items=items, lineage_items=lineage_items, ) - if watch_only_monitor_items: - summary["watch_only_monitor_count"] = len(watch_only_monitor_items) - summary["watch_only_monitor_due_count"] = len(watch_only_monitor_due_items) - summary["convergence_open_count"] = len(convergent_open_items) - if recent_completed_advancement_items: - summary["recent_completed_advancement_items"] = recent_completed_advancement_items if include_task_orchestration_authority: summary["task_orchestration_authority"] = _task_orchestration_authority( - lanes, - role=role, - ) - if lanes.blocker_items: - summary["blocker_open_count"] = len(lanes.blocker_items) - summary["blocker_items"] = [ - compact_todo_item(item) for item in lanes.blocker_items - ] - from ..effect_runtime import effect_runtime_result - - replan_gates = {gate.get("todo_id") for gate in handoff_gates - if gate.get("route_continuation_replan_required") is True} - closure = effect_runtime_result("todo.succession.closure", { - "schema_version": "todo_closure_request_v0", "role": role, - "source_section": source_section, "full_selection": full_selection, - "rows": [{"status": item.get("status") or ("done" if item.get("done") else "open"), - "watch_only": projection_todo_item_is_watch_only_monitor(item), - "no_followup": normalize_todo_no_followup(item.get("no_followup")) is True, - "successor_gap": evaluation["successor_gap"], "handoff_state": evaluation["handoff_state"], - "replan": item.get("route_continuation_replan_required") is True or item.get("todo_id") in replan_gates} - for item, evaluation in zip(items, succession, strict=True)], - }) - if not isinstance(closure, dict): - raise ValueError("invalid typed Todo closure projection") - summary.update(closure) - if lanes.resume_blocked_items: - summary["resume_blocked_count"] = len(lanes.resume_blocked_items) - summary["resume_blocked_items"] = [ - compact_todo_item(item) - for item in lanes.resume_blocked_items[:MAX_DEFERRED_TODO_VISIBILITY_ITEMS] - ] + projected["orchestration"], role=role) if handoff_gates: summary["handoff_gates"] = handoff_gates - if successor_gap_items: - compact_gap_items = successor_gap_items[:MAX_COMPLETED_SUCCESSION_WARNING_ITEMS] - summary["completed_without_successor_count"] = len(successor_gap_items) - summary["completed_without_successor_items"] = compact_gap_items + if summary.get("completed_without_successor_count"): summary["todo_succession_warning"] = { "schema_version": TODO_SUCCESSION_WARNING_SCHEMA_VERSION, "reason_code": TODO_SUCCESSION_WARNING_REASON_CODE, - "count": len(successor_gap_items), - "items": compact_gap_items, + "count": summary["completed_without_successor_count"], + "items": summary["completed_without_successor_items"], "recommended_action": ( "run loopx todo complete --no-follow-up for the completed Todo, " "or add/link a successor Todo before closing the slice; do not " "invent a user gate" ), } - if lanes.active_next_action_items: - summary["active_next_action_items"] = [ - compact_active_next_action_todo_item(item) - for item in lanes.active_next_action_items - ] - if lanes.active_next_action_executable_items: - summary["active_next_action_executable_items"] = [ - compact_active_next_action_todo_item(item) - for item in lanes.active_next_action_executable_items - ] - if lanes.claimed_open_items: - summary["claimed_open_count"] = len(lanes.claimed_open_items) - summary["unclaimed_open_count"] = len(lanes.open_items) - len(lanes.claimed_open_items) - summary["claimed_advancement_open_count"] = len(lanes.claimed_advancement_items) - summary["claimed_monitor_open_count"] = len(lanes.claimed_monitor_items) return summary diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index 4a51d08c9b..af3eabf1d2 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -1535,7 +1535,7 @@ }, { "site": "loopx/history.py::.collect_history::codec_read:load_registry#1", - "line": 301, + "line": 327, "column": 20, "kind": "codec_read", "api": "load_registry", @@ -1543,7 +1543,7 @@ }, { "site": "loopx/history.py::.inspect_index_duplicates::codec_read:load_registry#1", - "line": 542, + "line": 571, "column": 16, "kind": "codec_read", "api": "load_registry", @@ -1551,7 +1551,7 @@ }, { "site": "loopx/history.py::.rebuild_index_artifact_collisions::codec_read:load_registry#1", - "line": 756, + "line": 785, "column": 16, "kind": "codec_read", "api": "load_registry", @@ -1559,7 +1559,7 @@ }, { "site": "loopx/history.py::.repair_index_duplicates::codec_read:load_registry#1", - "line": 646, + "line": 675, "column": 16, "kind": "codec_read", "api": "load_registry", diff --git a/loopx/status.py b/loopx/status.py index 96290fa032..26ee24cf16 100644 --- a/loopx/status.py +++ b/loopx/status.py @@ -16,20 +16,20 @@ collect_status as _collect_status_read_model, ) from .control_plane.status.active_state_projection import ( - STATE_EVENT_LOG_BASENAME, + STATE_EVENT_LOG_BASENAME as STATE_EVENT_LOG_BASENAME, ) from .control_plane.status.contract_projection import ( - STATUS_CONTRACT_RELOAD_HINT, + STATUS_CONTRACT_RELOAD_HINT as STATUS_CONTRACT_RELOAD_HINT, ) from .control_plane.status.goal_attention_projection import ( - PLANNED_CONTROLLER_OPT_IN_RECOMMENDED_ACTION, + PLANNED_CONTROLLER_OPT_IN_RECOMMENDED_ACTION as PLANNED_CONTROLLER_OPT_IN_RECOMMENDED_ACTION, ) # Refs #4447: one definition for this vocabulary. The control_plane projection # owns it because it feeds the monitor/attention read models; this module keeps # re-exporting each name for existing callers instead of restating its value. from .control_plane.status.monitor_display_projection import ( - MONITOR_DISPLAY_FALLBACK_ACTION, - MONITOR_DISPLAY_STOP_CONDITION, + MONITOR_DISPLAY_FALLBACK_ACTION as MONITOR_DISPLAY_FALLBACK_ACTION, + MONITOR_DISPLAY_STOP_CONDITION as MONITOR_DISPLAY_STOP_CONDITION, MONITOR_SIGNAL_WAITING_ON, ) from .control_plane.status.registry_health_projection import ( @@ -189,7 +189,6 @@ active_state_todo_attention_item as _active_state_todo_attention_item_read_model, active_next_action_todo_ids, attach_dependency_blockers, - claimed_visibility_items as claimed_visibility_items, compact_todo_group as compact_todo_group, compact_todo_item as compact_todo_item, first_open_todo_text, @@ -234,7 +233,6 @@ "TODO_PROJECTION_DETAIL_POINTER_SCHEMA_VERSION": "loopx.control_plane.work_items.project_asset", "TODO_PROJECTION_VIEW_SCHEMA_VERSION": "loopx.control_plane.work_items.project_asset", "project_asset_summary_is_public_safe": "loopx.control_plane.work_items.project_asset", - "claimed_visibility_items": "loopx.control_plane.todos.todo_summary", "compact_todo_group": "loopx.control_plane.todos.todo_summary", "compact_todo_item": "loopx.control_plane.todos.todo_summary", "todo_lane_items": "loopx.control_plane.todos.todo_summary", diff --git a/tests/control_plane/test_todo_consumer_scope.py b/tests/control_plane/test_todo_consumer_scope.py index f8fbaa0a49..c5913db50d 100644 --- a/tests/control_plane/test_todo_consumer_scope.py +++ b/tests/control_plane/test_todo_consumer_scope.py @@ -97,16 +97,16 @@ def test_filter_composes_with_existing_lane_call_and_rejects_downgraded_response original = effect_runtime.effect_runtime_result requests = [] def track(method, request, **kwargs): - if method == "todo.summary_lanes.project": + if method == "todo.summary.project": requests.append(request) return original(method, request, **kwargs) monkeypatch.setattr(effect_runtime, "effect_runtime_result", track) assert filtered_todo_summary(source, role="user", agent_id="agent-a")["total_count"] == 1 assert len(requests) == 1 - assert requests[0]["schema_version"] == "todo_summary_lanes_request_v1" + assert requests[0]["schema_version"] == "todo_summary_projection_request_v0" def downgrade(method, request, **kwargs): result = original(method, request, **kwargs) - if method == "todo.summary_lanes.project": + if method == "todo.summary.project": result.pop("source_indices", None) return result monkeypatch.setattr(effect_runtime, "effect_runtime_result", downgrade) @@ -127,8 +127,8 @@ def test_typed_lane_response_cannot_alias_or_escape_selected_source(monkeypatch, original = effect_runtime.effect_runtime_result def corrupt(method, request, **kwargs): result = original(method, request, **kwargs) - if method == "todo.summary_lanes.project": - result["lanes"]["open_items"] = {"duplicate": [0, 0], "boolean": [False], "outside_selection": [outside]}[corruption] + if method == "todo.summary.project": + result["lanes"]["first_open_items"]["indices"] = {"duplicate": [0, 0], "boolean": [False], "outside_selection": [outside]}[corruption] return result monkeypatch.setattr(effect_runtime, "effect_runtime_result", corrupt) with pytest.raises(ValueError, match="source ordinal|escaped the selected source"): diff --git a/tests/control_plane/test_todo_summary_projection.py b/tests/control_plane/test_todo_summary_projection.py new file mode 100644 index 0000000000..bf706de776 --- /dev/null +++ b/tests/control_plane/test_todo_summary_projection.py @@ -0,0 +1,62 @@ +"""Whole-source summary decisions precede display limits and preserve chronology.""" +from loopx.control_plane.todos.todo_summary import compact_todo_group + + +def row(index, **fields): + return {"todo_id": f"todo_summary_{index}", "text": f"Work {index}", "role": "agent", + "task_class": "advancement_task", "status": "open", "index": index, + "source_section": "Agent Todo", **fields} + + +def summarize(items, **kwargs): + return compact_todo_group(items, role="agent", source_section="Agent Todo", **kwargs) + + +def test_recent_completions_compare_instants_not_offset_strings(): + items = [row(1, status="done", no_followup=True, completed_at="2026-01-01T10:00:00+08:00"), + row(2, status="done", no_followup=True, completed_at="2026-01-01T03:00:00Z")] + result = summarize(items) + assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == [ + "todo_summary_2", "todo_summary_1"] + + +def test_recent_completions_preserve_microseconds_across_offsets(): + items = [row(2, status="done", no_followup=True, completed_at="2026-01-01T10:00:00.000001+08:00"), + row(1, status="done", no_followup=True, completed_at="2026-01-01T02:00:00.000002Z")] + result = summarize(items) + assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == [ + "todo_summary_1", "todo_summary_2"] + + +def test_invalid_completion_time_does_not_displace_known_recent_work(): + result = summarize([row(1, status="done", no_followup=True, completed_at="unknown"), + row(2, status="done", no_followup=True, completed_at="2026-01-01T03:00:00Z")]) + assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == ["todo_summary_2"] + assert result["advancement_done_count"] == 2 # Still retained as completed work. + + +def test_summary_caps_do_not_change_work_counts_or_hide_a_peer(): + items = [row(i, claimed_by="agent-a" if i < 24 else "agent-b") for i in range(32)] + result = summarize(items, item_limit=1) + assert len(result["items"]) == 1 and result["work_counts"]["advancement"] == 32 + assert len(result["claimed_open_items"]) == 16 + assert {item["claimed_by"] for item in result["claimed_open_items"]} == {"agent-a", "agent-b"} + assert result["claimed_open_count"] == 32 + + +def test_late_edit_does_not_make_an_old_completion_recent(): + result = summarize([row(1, status="done", no_followup=True, + completed_at="2026-01-01T00:00:00Z", updated_at="2026-09-01T00:00:00Z"), + row(2, status="done", no_followup=True, + completed_at="2026-02-01T00:00:00Z")]) + assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == [ + "todo_summary_2", "todo_summary_1"] + + +def test_a_selection_cannot_restore_a_lost_full_source_proof(): + from loopx.control_plane.todos.todo_summary import compact_evaluated_todo_group + source = summarize([row(1, status="done", no_followup=True)], item_limit=None) + result = compact_evaluated_todo_group(source["items"], source_section="Agent Todo", role="agent", + full_selection=False, selection={"role": "agent", "status": None, "todo_id": None, "agent_id": None}) + assert "source_proof" not in result and "terminal_closure_proof" not in result + assert result["done_count"] == 1 diff --git a/tests/control_plane_ts/todo_consumer_scope_conformance.ts b/tests/control_plane_ts/todo_consumer_scope_conformance.ts index 19e5754403..c44c08f2d3 100644 --- a/tests/control_plane_ts/todo_consumer_scope_conformance.ts +++ b/tests/control_plane_ts/todo_consumer_scope_conformance.ts @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import {spawnSync} from "node:child_process"; import test from "node:test"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts"; import {productionScaleConsumerScopeFixture} from "./production_scale_coordination_fixture.ts"; @@ -24,6 +25,7 @@ prefix=lambda values: sorted(row['todo_id'] for row in values if row['todo_id']. result={'selected':prefix(selected['items']), 'quota_delta':quota['open_count']-base_quota['open_count'], 'limited':len(limited['items']), 'counts_equal':selected['total_count']==limited['total_count'], 'whole':prefix(summary['items']), + 'recent':[row['todo_id'] for row in fields['agent_todos'].get('recent_completed_advancement_items', [])[:2]], 'filtered_peer':filtered_todo_summary(summary,role='user',agent_id='agent-a',todo_id='todo_scope_peer_gate')['items'], 'succession_gap':filtered_todo_summary(fields['agent_todos'],role='agent',todo_id=p['cases']['inferred_source']).get('completed_without_successor_count',0)} print(json.dumps(result)) @@ -32,6 +34,12 @@ export function registerTodoConsumerScopeConformance(name: string, factory: Auth for (const schema of ["native", "legacy"] as const) test(`${name}: full-source Agent read addressing (${schema})`, async context => { const {store} = await factory(context); const {projection, cases} = productionScaleConsumerScopeFixture("consumer-scope", schema); + const recent = (projection.todos as JsonObject[]).filter(todo => todo.status === "done" && + todo.task_class === "advancement_task" && todo.archive_state !== "archive").slice(0, 2); + assert.equal(recent.length, 2); + recent[0].completed_at = "2099-01-01T10:00:00.000001+08:00"; + recent[0].updated_at = "2099-12-01T00:00:00Z"; + recent[1].completed_at = "2099-01-01T02:00:00.000002Z"; assert.equal((await store.commitAuthority({operation_id: "scope-source", expected_provider_revision: null, next_projection: projection, events: [], receipts: []})).status, "applied"); const before = await store.loadAuthority(); assert.equal(before.status, "loaded"); @@ -45,6 +53,7 @@ export function registerTodoConsumerScopeConformance(name: string, factory: Auth assert.equal(result.whole.length, 5); assert.deepEqual(result.filtered_peer, []); assert.equal(result.limited, 1); assert.equal(result.counts_equal, true); assert.equal(result.succession_gap, 0); + assert.deepEqual(result.recent, [recent[1].todo_id, recent[0].todo_id]); assert.deepEqual(await store.loadAuthority(), before, "read consumers must never mutate authority"); }); } diff --git a/tests/control_plane_ts/todo_summary_projection.test.ts b/tests/control_plane_ts/todo_summary_projection.test.ts new file mode 100644 index 0000000000..1417bf213f --- /dev/null +++ b/tests/control_plane_ts/todo_summary_projection.test.ts @@ -0,0 +1,102 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; +import {projectTodoSummary} from "../../loopx/control_plane/todos/summary_projection.ts"; +import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts"; + +const row = (fields: JsonObject = {}): JsonObject => ({status: "open", done: false, + task_class: "advancement_task", has_resume: false, resume_ready: null, resume_evaluated: false, + acceptance_blocked: false, claimed: false, claim: null, preferred: false, watch_only: false, + due_at: null, expires_at: null, sort: [1, 1, "", ""], todo_id: null, bound: null, + blocks: null, global: false, excluded: [], completed_at: null, updated_at: null, + completion_index: 0, linked_user_action: false, no_followup: false, + successor_gap: false, handoff_state: null, replan: false, ...fields}); +const request = (rows: JsonObject[], fields: JsonObject = {}): JsonObject => ({ + schema_version: "todo_summary_projection_request_v0", rows, observed_at: 100, + selection: null, role: "agent", source_section: "Agent Todo", item_limit: 12, full_selection: true, ...fields}); +const done = (fields: JsonObject = {}) => row({status: "done", done: true, no_followup: true, ...fields}); + +test("one whole-source projection computes counts, visibility and closure before limits", () => { + for (const limit of [null, 0, 1, 12]) { + const rows = Array.from({length: 32}, (_, index) => row({claimed: true, claim: index < 24 ? "a" : "b"})); + const before = structuredClone(rows), result = projectTodoSummary(request(rows, {item_limit: limit})); + assert.equal(result.fields.open_count, 32); + assert.equal((result.fields.work_counts as JsonObject).advancement, 32); + assert.equal(result.fields.claimed_open_count, 32); + assert.deepEqual(result.lanes.claimed_open_items.indices, [...Array(8).keys(), ...Array.from({length: 8}, (_, i) => 24 + i)]); + assert.equal(result.lanes.items.indices.length, limit === null ? 32 : limit); + assert.equal(result.fields.terminal_closure_proof, undefined); + assert.deepEqual(rows, before); + } +}); + +test("recent completion orders true microsecond instants and ignores later edits", () => { + const rows = [done({completed_at: "2026-01-01T10:00:00.000001+08:00", updated_at: "2026-12-01T00:00:00Z"}), + done({completed_at: "2026-01-01T02:00:00.000002Z"}), done({completed_at: "invalid"})]; + const result = projectTodoSummary(request(rows)); + assert.deepEqual(result.lanes.recent_completed_advancement_items.indices, [1, 0]); + assert.equal(result.fields.advancement_done_count, 3); + assert.equal(result.lanes.items.indices.length, 3); +}); + +test("equal instants retain reverse source coordinate and stable ties", () => { + const rows = [done({completed_at: "2026-01-01T10:00:00+08:00", completion_index: 2}), + done({completed_at: "2026-01-01T02:00:00Z", completion_index: 4}), + done({completed_at: "2026-01-01T02:00:00Z", completion_index: 4})]; + assert.deepEqual(projectTodoSummary(request(rows)).lanes.recent_completed_advancement_items.indices, [1, 2, 0]); +}); + +test("selection cannot turn partial source knowledge into a closure proof", () => { + const select = {role: "agent", status: null, todo_id: null, agent_id: null}; + const full = projectTodoSummary(request([done()], {selection: select})); + assert.ok(full.fields.terminal_closure_proof); + const partial = projectTodoSummary(request([done()], {selection: select, full_selection: false})); + assert.equal(partial.full_selection, false); + assert.equal(partial.fields.source_proof, undefined); + assert.equal(partial.fields.terminal_closure_proof, undefined); + const filtered = projectTodoSummary(request([done(), row()], {selection: {...select, status: "done"}})); + assert.equal(filtered.fields.terminal_closure_proof, undefined); + assert.deepEqual(filtered.source_indices, [0]); +}); + +test("scope preserves original ordinals and closure uses only the selected graph decisions", () => { + const rows = [row({todo_id: "peer", claimed: true, claim: "other"}), + done({todo_id: "own", claimed: true, claim: "me"})]; + const result = projectTodoSummary(request(rows, {selection: {role: "agent", status: null, todo_id: null, agent_id: "me"}})); + assert.deepEqual(result.source_indices, [1]); + assert.deepEqual(result.lanes.items.indices, [1]); + assert.equal(result.fields.done_count, 1); + assert.equal(result.fields.terminal_closure_proof, undefined); +}); + +test("invalid source or budgets fail closed rather than hiding rows", () => { + for (const fields of [{item_limit: -1}, {item_limit: 1.5}, {item_limit: true}, {full_selection: null}]) { + assert.throws(() => projectTodoSummary(request([row()], fields))); + } + for (const fields of [{claimed: true}, {claim: "agent"}, {done: true}, {completed_at: 2}, + {has_resume: true}, {successor_gap: "false"}, {handoff_state: "unrecognized"}]) { + assert.throws(() => projectTodoSummary(request([row(fields)]))); + } +}); + +test("large native and imported corpora preserve source coverage across every display cap", () => { + for (const format of ["native", "legacy"] as const) { + const fixture = productionScaleCoordinationFixture("summary-source", format); + const records = fixture.projection.todos as JsonObject[]; + const rows = records.map((todo, ordinal) => row({status: todo.status, done: todo.done, + task_class: todo.task_class, sort: [1, ordinal, "", ""], todo_id: todo.todo_id, + claim: todo.claimed_by ?? null, claimed: Boolean(todo.claimed_by)})); + const full = projectTodoSummary(request(rows, {item_limit: null})); + assert.equal(full.fields.total_count, records.length); + assert.equal(full.lanes.items.indices.length, records.length); + for (const limit of [0, 1, 12]) { + const limited = projectTodoSummary(request(rows, {item_limit: limit})); + assert.deepEqual(limited.fields, full.fields); + assert.equal(limited.lanes.items.indices.length, limit); + for (const lane of Object.values(limited.lanes)) { + assert.equal(new Set(lane.indices).size, lane.indices.length); + assert.ok(lane.indices.every(index => index >= 0 && index < records.length)); + } + } + } +}); From 4c327833f9b88203ba5f3d97f197a3aa89bfd365 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 02:00:43 +0800 Subject: [PATCH 2/3] docs(rfc): reconcile summary ownership and local-default dependencies Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...shared-goal-authority-state-provider-v0.md | 18 ++++++- ...-goal-authority-state-provider-v0.zh-CN.md | 13 +++++ .../typescript-control-plane-migration-v0.md | 13 +++++ ...script-control-plane-migration-v0.zh-CN.md | 10 ++++ docs/reference/todo-work-counts.md | 54 +++++++++++++++++-- 5 files changed, 103 insertions(+), 5 deletions(-) diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md index dd1247ca0a..6a87896a6b 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -3110,7 +3110,15 @@ This closes that T3/L5 consumer family, not D1 permanent display freshness, D2 durability, D3 whole-Goal qualification or default-provider selection. The conditional 5–8 remaining delivery-package estimate is unchanged. -Summary/work-lane counts now remain independent of display limits and retain incomplete-source knowledge through Agent scoping; canonical list acceptance holds match status. This closes one L5 read consumer, not permanent projection freshness or D1–D3. See [count semantics](../../reference/todo-work-counts.md). +The Todo summary consumer now uses one TS batch for scope, lanes, counts, +claimant-balanced display and closure. It retires Python count/cap/allocation +branches and two separate internal lane/closure calls. Recent completion uses +actual completion instants, not edit time or ISO-string order; partial source +knowledge cannot become a whole-source closure proof after selection. Public +summary/persisted record schemas and display budgets stay unchanged. Real CLI +and complete-graph provider readback qualify this read-model boundary, not +permanent projection freshness or all D1–D3. See +[count and chronology semantics](../../reference/todo-work-counts.md). The Goal Channel ownership observation consumes one complete provider revision before bounding display. It never repairs Markdown or revives old local leases; provider failures and truncation stay visible. This is a T3 read closure with shared TS interpretation, not D1/D2 qualification or D3 cutover. See [coordination observation](../../reference/coordination-observation.md). @@ -3281,6 +3289,14 @@ PRs**, conditional on the caller audit finding no additional missing effects: | L7 capture plus L8 integrated migration | 1–2 | Mixed-writer continuity, fenced whole-Goal rehearsal, export/rollback and cohort evidence. | | L9 default and bounded retirement | 1 | New-Goal onboarding/settings/install choose the qualified profile; remove final obsolete callers. | +At the 2026-09-24 reconciliation, #4870/#4888/#4920 are merged; +#4922 (snapshot pagination), #4931 (SQLite proof encoding), #4960 (qualified +SQLite runtime admission) and #4961 (refresh display recovery) remain separate +in-review dependencies. This summary slice closes shared read-model decisions, +not those delivery gates. Requalify the combined accepted head before reducing +the conditional **5–8 package** estimate; do not count each helper migration as +one complete package. SQLite #4224 retains failing/missing D2 evidence. + The command-observation/current-proof closure removes a concrete L2/L3 concurrency hold. The retained-Monitor cycle and grouped executor closure remove concrete L4 holds, not an entire remaining package: the **5–8 PR planning range remains conditional**, rather than diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index 04b513b131..926360f54b 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -2421,6 +2421,14 @@ Scoped fallback 的选择与门禁关系也已复用同一 TS decision owner， **D1 — 资格化永久投影交付，可与 T1/T2 重叠推进。** +Todo 摘要现由一个 TS 批次决定范围、lanes、计数、claim 展示分配和收尾证明，删除 +Python 的重复汇总分支及分离的 lane／closure 内部调用。最近完成按真实完成时刻排序， +不把编辑时间或 ISO 字符串顺序当作完成顺序；partial 来源不能在再次筛选后重新取得 +整源收尾证明。公开摘要／持久记录合同与展示预算保持不变。 +见[计数与时间语义](../../reference/todo-work-counts.md)。这关闭共享摘要决策，不宣称 +全部消费者、永久展示新鲜度或 D1–D3 已通过。 + + 摘要与 work-lane 计数已独立于展示上限，并在 Agent 筛选后保留来源不完整状态；canonical 列表的 acceptance 限制与 status 一致。这只闭合 L5 的一个读取消费者，不代表永久投影新鲜度或 D1–D3 完成。见[计数语义](../../reference/todo-work-counts.md)。 Goal Channel 所有权观察先读取完整 provider revision，再限制展示；不修复 Markdown、不复活旧本地 lease，明确披露失败与截断。这是共用 TS 解释规则的 T3 读链路闭合，不完成 D1/D2 或 D3 切换，见 [coordination observation](../../reference/coordination-observation.md)。 @@ -2532,6 +2540,11 @@ D1 交付确认现于 Markdown 耐久读回后核对 canonical revision。未固 | C／L8：整 Goal 演练与分组迁移 | L2–L7 后汇合一个精确 revision／profile；drain capture、fence 旧 writer、回读 canonical 与投影、演练 fenced export／rollback。 | D3 包绑定 lineage、cursor、source digest、命令覆盖和 profile；已有 Goal 分组迁移需明确批准，不能按命令拆 authority 或复活旧 Markdown。 | | D／L9：新 Goal 默认与有界退役 | 单独 default-change PR 让新建／onboarding 选择合格本地 profile，配齐 settings／readback、installer 和打包客户端；最后 caller 与迁移窗口退出才删除旧业务 writer。 | L8 整体产品／回滚资格；区分新 Goal 默认和已有 Goal 迁移。发布兼容／停用说明，保留显式 provider、永久 renderer 和合法 import/export。T4 可在默认启用后继续收尾。 | +**2026-09-24 依赖核对。** #4870/#4888/#4920 已合并；#4922 快照分页、#4931 +SQLite 证明编码优化、#4960 SQLite runtime 准入和 #4961 刷新显示恢复仍属独立在评审 +依赖。#4224 的 D2 失败／缺项仍需关闭；本次摘要规则收口不替代它们。以下条件式 +**5–8 个交付包**需在组合 head 验收后更新，不能按 helper 迁移数量机械扣减。 + **开发节奏以证据推进。** 先核对在途 stack，再按完整操作交付 A；L6/L7 可独立推进。 B 汇合为完整用户流程，C 形成一次可复现资格检查点，D 用独立 PR 修改默认。 按当前已合并边界，剩余 caller／executor 约 1–2 个包，consumer／投影 1 个， diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 5ec9d57164..a3fb0ef817 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1888,3 +1888,16 @@ This is a migration orchestration checkpoint, not completion of Stage 3 or a default-provider flip. Integrate claim-preserving migration separately, retain real-backend and captured-source qualification, and retire Python only where its actual callers have moved. [Operator contract](../../reference/reviewed-coordination-promotion.md). + + +### Todo summary decision ownership + +One TS summary batch now owns selected-source counts, display allocation, +recent-completion chronology, orchestration candidate positions and closure +proofs. Python retains decoding, public field allowlists and rendering. The +old Python claimant selector and aggregate branches are retired; the internal +lane and closure RPC entries are replaced without retaining unused wire paths. +Public `todo_summary_v0` and persisted records do not change. Full-source +relationship evaluation is reused before selection, and source completeness is +preserved independently of query matching. See [semantics and rollback](../../reference/todo-work-counts.md). +This advances T3/L5; it does not replace D2/D3 or flip a provider default. 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 fa8ef46bd9..2fe9e081da 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 @@ -1449,3 +1449,13 @@ This is a migration orchestration checkpoint, not completion of Stage 3 or a default-provider flip. Integrate claim-preserving migration separately, retain real-backend and captured-source qualification, and retire Python only where its actual callers have moved. [Operator contract](../../reference/reviewed-coordination-promotion.md). + + +### Todo 摘要决策收口 + +已选来源的计数、展示分配、最近完成时间顺序、编排候选位置与收尾证明，收口到一个 +TS 摘要批次；Python 保留旧格式解码、公开字段筛选及渲染。删除旧 Python claim 分配 +算法和汇总分支，用一个内部入口替换 lane／closure 两次调用，不保留无调用方的旧 wire。 +公开 `todo_summary_v0` 和持久记录不变；完整来源的关系求值先于筛选，来源完整性不被 +查询命中情况覆盖。见[语义及回滚](../../reference/todo-work-counts.md)。这是 T3/L5 的 +共享读取边界推进，不替代 D2/D3 或 provider 默认切换。 diff --git a/docs/reference/todo-work-counts.md b/docs/reference/todo-work-counts.md index 27ebe38343..6d0386a631 100644 --- a/docs/reference/todo-work-counts.md +++ b/docs/reference/todo-work-counts.md @@ -79,10 +79,11 @@ read and continue to show the whole Goal; no new configuration editor is needed. Resume and succession are evaluated on the complete source before selection. The typed batch filters rows without renumbering their original source indexes, then builds lanes/counts, and only then applies display limits. Status/identity -filters do not recompute dependencies from their smaller view. The v1 internal -request composes this selection into the existing call; v0 unfiltered callers -retain their wire contract. Python decodes legacy input and renders results, -with no independent Agent-addressing rule. +filters do not recompute dependencies from their smaller view. One internal `todo.summary.project` batch now composes selection, counts, +visibility allocation and closure. Its transient request replaces the separate +lane and closure RPC calls; it does not change persisted Todo or public summary +schemas. Python decodes legacy input, validates source ordinals and materializes +public fields, with no independent summary count, cap or claimant-allocation rule. ## 中文说明 @@ -112,3 +113,48 @@ claim/exclusion 筛选，可见不代表获准执行。 当前列表。未筛选的整 Goal 视图仍显示这些记录。依赖和 succession 先在完整来源求值， TS 再筛选并保留原数组位置，最后生成 lanes、计数和有界展示；筛选后的数组位置不是原 来源位置。无需新增 capability、配置、前端或 Lark 编辑入口，不增加一次筛选 RPC。 + +## Summary chronology and source completeness + +The same TS projection now supplies `recent_completed_advancement_items`, +claimant-balanced display lanes, orchestration candidate positions and closure +proofs to legacy and canonical consumers. Display budgets are unchanged; +`items` limits never change full-source counts. Full-source resume/succession +evaluation still precedes filtering, and returned positions refer to the original +array, not a newly numbered subset. Python retains public field allowlists, +warning text, privacy redaction and Markdown parsing/rendering. + +**Intentional read behavior changes:** recent completions are ordered by the +actual `completed_at` instant, preserving timezone offsets and microseconds. +Later `updated_at` edits no longer make an old completion recent. Missing or +invalid completion times remain in completed-work counts/history but do not +claim a place in the recent-completion lane. Equal instants retain reverse +source-coordinate order and stable ties. Succession warnings keep their +last-change ordering, now comparing instants rather than timestamp strings; +unknown instants follow known ones without discarding the warning. + +A source already marked partial cannot regain `source_proof` or +`terminal_closure_proof` simply because a later selection matches all visible +rows. Query scope and source completeness are independent conditions. These +proofs remain read-only observations, not permission to settle a Goal. + +This changes status, Todo-list and quota summary readback for both legacy and +promoted Goals without a flag. Existing frontend and Lark views consume these +Core projections; no new setting or frontend asset is required. No provider, +lease, registry or display writer is added. Rollback requires the matching +Python/TS package but no data migration. Full L5 consumer acceptance, projection +freshness, SQLite D2 and default/cutover gates remain separate. + +### 中文补充 + +摘要的计数、展示上限、领取者之间的展示分配、编排候选位置及收尾证明，现由一个 TS +批次决定；删除 Python 的重复汇总分支和仅为旧内部调用保留的 claim 分配 helper。 +Python 继续负责旧数据解码、公开字段筛选、隐私处理与文本展示。 + +这是有意的读取语义修复：最近完成列表按 `completed_at` 的真实时刻排序，保留时区和 +微秒，不再把较晚编辑误作较晚完成。缺失／非法时间仍计入已完成总数和历史，但不进入 +最近完成列表。后继缺口警告仍按最后更新时间排序，未知时间靠后，不丢弃警告。 +已有 partial 来源不会因为再次筛选命中所有可见行，就重新获得整个来源的收尾证明。 + +覆盖 legacy 与 canonical 的 status、Todo 查询和 quota 摘要；展示预算保持原值。 +没有新增设置、权限或 writer，不改变 provider 默认值，也不宣称完成整 Goal 迁移。 From 0a30a25aa6e2656db27a054442393ec21a8cd130 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 14:26:23 +0800 Subject: [PATCH 3/3] fix(todos): keep display failure semantics and long-history batches inside budget Display may never invent the source's resume decision, so the evaluator that the row facts already compute is now asserted before the succession RPC: an unevaluated source fails with the source's own "full-source resume evaluation" diagnostic instead of a later owner's generic "succession evaluation must be an object". Five negative variants and the completed-history HTTP read show it. That HTTP read also exposed the second defect: one whole-source batch repeated every key name per Todo, so 4087 completed Todos produced a 2.05 MiB request and failed closed with a 400. The adapter now sends the declared columnar facts and the typed owner decodes them back into the same row objects before validating, which brings the same source to about 0.73 MiB. Nothing is defaulted from absence, and the column order is validated as this request version's schema. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../control_plane/todos/summary_projection.ts | 34 +++++++++++- loopx/control_plane/todos/todo_summary.py | 54 ++++++++++++++----- .../control_plane/test_todo_consumer_scope.py | 6 ++- .../test_todo_summary_projection.py | 24 +++++++++ .../todo_summary_projection.test.ts | 7 ++- 5 files changed, 108 insertions(+), 17 deletions(-) diff --git a/loopx/control_plane/todos/summary_projection.ts b/loopx/control_plane/todos/summary_projection.ts index 7dbf0a7e36..f0c3af39e2 100644 --- a/loopx/control_plane/todos/summary_projection.ts +++ b/loopx/control_plane/todos/summary_projection.ts @@ -18,6 +18,36 @@ interface SummaryProjection { orchestration: {candidate_items: number[]; user_blocker_items: number[]}; } +/** The declared order is this request version's schema, not a hint: the same + * cells in a different order would silently change meaning. */ +export const TODO_SUMMARY_PROJECTION_COLUMNS = [ + "status", "done", "task_class", "has_resume", "resume_ready", "resume_evaluated", + "acceptance_blocked", "claimed", "preferred", "watch_only", "due_at", "expires_at", + "sort", "completed_at", "updated_at", "completion_index", "linked_user_action", + "no_followup", "successor_gap", "handoff_state", "replan", "todo_id", "claim", + "bound", "blocks", "global", "excluded", +] as const; + +/** One whole-source batch carries every Todo, so the co-deployed adapter sends + * columnar facts. Decoding restores the row objects the lane and closure owners + * already validate; nothing is defaulted or inferred from absence. */ +function decodeRows(request: JsonObject): JsonObject[] { + const columns = request.columns; + if (!Array.isArray(columns) || columns.length !== TODO_SUMMARY_PROJECTION_COLUMNS.length || + columns.some((name, index) => name !== TODO_SUMMARY_PROJECTION_COLUMNS[index])) { + throw new EffectRuntimeRequestError("Todo summary row columns do not match the typed adapter order"); + } + if (!Array.isArray(request.rows)) { + throw new EffectRuntimeRequestError("Todo summary rows must be a list"); + } + return request.rows.map((value, ordinal) => { + if (!Array.isArray(value) || value.length !== columns.length) { + throw new EffectRuntimeRequestError(`Todo summary row ${ordinal} does not match its declared columns`); + } + return Object.fromEntries(TODO_SUMMARY_PROJECTION_COLUMNS.map((name, index) => [name, value[index]])); + }); +} + /** Allocate a bounded display across claimants, then restore source ordering. */ function claimedVisibility(indices: readonly number[], rows: readonly JsonObject[], limit: number): number[] { if (indices.length <= limit) return [...indices]; @@ -42,7 +72,7 @@ function claimedVisibility(indices: readonly number[], rows: readonly JsonObject export function projectTodoSummary(value: unknown): SummaryProjection { const request = requireJsonObject(value, "Todo summary request"); - if (request.schema_version !== "todo_summary_projection_request_v0" || !Array.isArray(request.rows)) { + if (request.schema_version !== "todo_summary_projection_request_v1") { throw new EffectRuntimeRequestError("Todo summary request schema mismatch"); } const role = request.role === null ? null : requireStringLiteral(request.role, ["user", "agent"], "role"); @@ -54,7 +84,7 @@ export function projectTodoSummary(value: unknown): SummaryProjection { throw new EffectRuntimeRequestError("item_limit must be a non-negative integer or null"); } const full = requireBoolean(request.full_selection, "full_selection"); - const rows = request.rows.map(value => requireJsonObject(value, "summary row")); + const rows = decodeRows(request); // The co-deployed adapter sends source facts, not prose or full Todo bodies. for (const row of rows) { if ((row.claim !== null && typeof row.claim !== "string") || row.claimed !== Boolean(row.claim)) { diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py index 9f6cbc9763..5f739d34de 100644 --- a/loopx/control_plane/todos/todo_summary.py +++ b/loopx/control_plane/todos/todo_summary.py @@ -2,7 +2,7 @@ from datetime import datetime import re -from typing import Any, Callable, Optional +from typing import Any, Callable, Optional, TypeGuard from ..goals.goal_vision_wait_projection import attach_active_vision_waits from .contract import ( @@ -85,6 +85,19 @@ TASK_ORCHESTRATION_AUTHORITY_SCHEMA_VERSION = "task_orchestration_authority_v0" TODO_ARCHIVE_STATE_ACTIVE = "active" + +# One internal batch carries the whole source, so the adapter sends columnar +# facts: repeating every key name per Todo pushed a long-history request past the +# effect-runtime request budget. The typed owner decodes the declared columns +# back into row objects before validating them, so no cell changes meaning. +SUMMARY_PROJECTION_REQUEST_SCHEMA_VERSION = "todo_summary_projection_request_v1" +SUMMARY_PROJECTION_COLUMNS = ( + "status", "done", "task_class", "has_resume", "resume_ready", "resume_evaluated", + "acceptance_blocked", "claimed", "preferred", "watch_only", "due_at", "expires_at", + "sort", "completed_at", "updated_at", "completion_index", "linked_user_action", + "no_followup", "successor_gap", "handoff_state", "replan", "todo_id", "claim", + "bound", "blocks", "global", "excluded", +) AttentionItemBuilder = Callable[..., dict[str, Any]] GoalLifecycleFields = Callable[[dict[str, Any], Optional[dict[str, Any]]], dict[str, Any]] PublicSafeText = Callable[..., Optional[str]] @@ -805,6 +818,16 @@ def _structured_resume_source_items( ] +def _resume_condition_evaluated(item: dict[str, Any], resume: str | None) -> bool: + """The source's own full-source resume evaluation for this condition.""" + condition = item.get("resume_condition") + return (isinstance(condition, dict) + and condition.get("schema_version") == "todo_resume_condition_v0" + and condition.get("resume_when") == resume + and isinstance(condition.get("satisfied"), bool) + and item.get("resume_ready") is condition.get("satisfied")) + + def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None, *, selection: dict[str, Any] | None, role: str | None, source_section: str | None, item_limit: int | None, full_selection: bool, @@ -814,6 +837,14 @@ def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] | from .succession_warning import project_succession + # Display may never invent the source's resume decision. Assert the + # full-source precondition before any RPC that reuses the evaluation, so an + # unevaluated source fails with its own diagnostic instead of a downstream + # "succession evaluation must be an object" from a later owner. + for item in items: + resume = normalize_todo_resume_when(item.get("resume_when")) + if resume and not _resume_condition_evaluated(item, resume): + raise ValueError("Todo display requires a matching full-source resume evaluation") succession = project_succession(items, reuse=True) handoff_gates = build_todo_handoff_gate_states(items, evaluations=succession) replan_gates = {gate.get("todo_id") for gate in handoff_gates @@ -821,12 +852,7 @@ def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] | rows = [] for item, evaluation in zip(items, succession, strict=True): resume = normalize_todo_resume_when(item.get("resume_when")) - condition = item.get("resume_condition") - evaluated = (isinstance(condition, dict) - and condition.get("schema_version") == "todo_resume_condition_v0" - and condition.get("resume_when") == resume - and isinstance(condition.get("satisfied"), bool) - and item.get("resume_ready") is condition.get("satisfied")) + evaluated = _resume_condition_evaluated(item, resume) due = projection_todo_item_next_due_at(item) expires = projection_todo_item_expires_at(item) guard = item.get("goal_acceptance_guard") @@ -853,8 +879,10 @@ def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] | "excluded": normalize_todo_excluded_agents(item.get("excluded_agents"))}}) try: result = effect_runtime_result("todo.summary.project", { - "schema_version": "todo_summary_projection_request_v0", - "rows": rows, "observed_at": now_utc().timestamp(), + "schema_version": SUMMARY_PROJECTION_REQUEST_SCHEMA_VERSION, + "columns": list(SUMMARY_PROJECTION_COLUMNS), + "rows": [[row[name] for name in SUMMARY_PROJECTION_COLUMNS] for row in rows], + "observed_at": now_utc().timestamp(), "selection": selection, "role": role, "source_section": source_section, "item_limit": item_limit, "full_selection": full_selection, }) @@ -863,13 +891,15 @@ def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] | if not isinstance(result, dict) or result.get("schema_version") != "todo_summary_projection_v0": raise ValueError("invalid typed Todo summary projection") - def valid_ordinals(value: Any) -> bool: + def valid_ordinals(value: Any) -> TypeGuard[list[int]]: return (isinstance(value, list) and all(type(index) is int and 0 <= index < len(items) for index in value) and len(set(value)) == len(value)) selected = result.get("source_indices") - if not valid_ordinals(selected) or type(result.get("full_selection")) is not bool: + if not valid_ordinals(selected): + raise ValueError("invalid typed Todo selection ordinals") + if type(result.get("full_selection")) is not bool: raise ValueError("invalid typed Todo selection ordinals") selected_set = set(selected) lanes, orchestration = result.get("lanes"), result.get("orchestration") @@ -982,7 +1012,7 @@ def compact_evaluated_todo_group( items = projected["items"] if not items and not include_empty_source: return None - summary = projected["summary"] + summary: dict[str, Any] = projected["summary"] handoff_gates = build_todo_handoff_gate_states(items, evaluations=projected["succession"]) attach_advancement_frontier_revision_index(summary, items, role=role) attach_active_vision_waits( diff --git a/tests/control_plane/test_todo_consumer_scope.py b/tests/control_plane/test_todo_consumer_scope.py index c5913db50d..1bae6b3aa5 100644 --- a/tests/control_plane/test_todo_consumer_scope.py +++ b/tests/control_plane/test_todo_consumer_scope.py @@ -103,7 +103,11 @@ def track(method, request, **kwargs): monkeypatch.setattr(effect_runtime, "effect_runtime_result", track) assert filtered_todo_summary(source, role="user", agent_id="agent-a")["total_count"] == 1 assert len(requests) == 1 - assert requests[0]["schema_version"] == "todo_summary_projection_request_v0" + assert requests[0]["schema_version"] == "todo_summary_projection_request_v1" + # One whole-source batch stays columnar, so adding fields cannot silently + # push a long-history request past the runtime request budget. + assert requests[0]["columns"][:3] == ["status", "done", "task_class"] + assert all(len(cells) == len(requests[0]["columns"]) for cells in requests[0]["rows"]) def downgrade(method, request, **kwargs): result = original(method, request, **kwargs) if method == "todo.summary.project": diff --git a/tests/control_plane/test_todo_summary_projection.py b/tests/control_plane/test_todo_summary_projection.py index bf706de776..0eac0bfc2e 100644 --- a/tests/control_plane/test_todo_summary_projection.py +++ b/tests/control_plane/test_todo_summary_projection.py @@ -1,4 +1,6 @@ """Whole-source summary decisions precede display limits and preserve chronology.""" +import json + from loopx.control_plane.todos.todo_summary import compact_todo_group @@ -60,3 +62,25 @@ def test_a_selection_cannot_restore_a_lost_full_source_proof(): full_selection=False, selection={"role": "agent", "status": None, "todo_id": None, "agent_id": None}) assert "source_proof" not in result and "terminal_closure_proof" not in result assert result["done_count"] == 1 + + +def test_long_history_stays_inside_the_runtime_request_budget(monkeypatch): + """A whole-source batch must not outgrow the co-deployed runtime's request.""" + from loopx.control_plane import effect_runtime + from loopx.control_plane.effect_runtime import MAX_REQUEST_BYTES + requests = [] + original = effect_runtime.effect_runtime_result + + def track(method, request, **kwargs): + if method == "todo.summary.project": + requests.append(request) + return original(method, request, **kwargs) + + monkeypatch.setattr(effect_runtime, "effect_runtime_result", track) + items = [row(index, status="done", no_followup=True, + completed_at="2026-01-01T00:00:00Z") for index in range(4096)] + result = summarize(items, item_limit=None) + assert result["done_count"] == 4096 + request = requests[0] + encoded = json.dumps(request, separators=(",", ":")).encode() + assert len(encoded) < MAX_REQUEST_BYTES diff --git a/tests/control_plane_ts/todo_summary_projection.test.ts b/tests/control_plane_ts/todo_summary_projection.test.ts index 1417bf213f..53e4c91472 100644 --- a/tests/control_plane_ts/todo_summary_projection.test.ts +++ b/tests/control_plane_ts/todo_summary_projection.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; -import {projectTodoSummary} from "../../loopx/control_plane/todos/summary_projection.ts"; +import {TODO_SUMMARY_PROJECTION_COLUMNS, projectTodoSummary} from "../../loopx/control_plane/todos/summary_projection.ts"; import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts"; const row = (fields: JsonObject = {}): JsonObject => ({status: "open", done: false, @@ -12,7 +12,10 @@ const row = (fields: JsonObject = {}): JsonObject => ({status: "open", done: fal completion_index: 0, linked_user_action: false, no_followup: false, successor_gap: false, handoff_state: null, replan: false, ...fields}); const request = (rows: JsonObject[], fields: JsonObject = {}): JsonObject => ({ - schema_version: "todo_summary_projection_request_v0", rows, observed_at: 100, + schema_version: "todo_summary_projection_request_v1", + columns: [...TODO_SUMMARY_PROJECTION_COLUMNS], + rows: rows.map(row => TODO_SUMMARY_PROJECTION_COLUMNS.map(name => row[name] ?? null)), + observed_at: 100, selection: null, role: "agent", source_section: "Agent Todo", item_limit: 12, full_selection: true, ...fields}); const done = (fields: JsonObject = {}) => row({status: "done", done: true, no_followup: true, ...fields});