diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.md new file mode 100644 index 0000000000..05dd1b459f --- /dev/null +++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.md @@ -0,0 +1,99 @@ +# Goal-route hints in the existing planning batch + +## Boundary and compatibility + +This slice moves the complete read-only Goal-route projection from Python into +`work_items/goal_route_hint.ts`, consumed by the existing +`work_item.action_portfolio.project` batch. It retires the Python route builder +and task-override composer together with the production caller switch. The +existing `quota.run_decision` owner still produces task-orchestration overrides; +this slice consumes that result rather than implementing its policy again. + +The internal `quota_planning_packet_request_v1` carries independently nullable +planning and route contexts. Its result is `quota_planning_packet_v1`. The +coordination contract JSON registers both identities, and the existing generator +owns Python/TypeScript constants. Existing v0 requirements and result shape keep +their meaning. The external nested `goal_route_hint_v0` does not change. + +Python captures route facts at the former builder position, after Agent lane, +frontier and recommendation selection. The existing planning call projects the +hint before monitor-only, unadmitted-selection, settlement and cadence cleanup. +There is no later scheduler/serialization refill. Source list absence remains +separate from an authoritative empty list; an empty frontier object remains a +frontier. Candidate order, selection-before-display truncation, local-source +counts, current-action priority, canonical frontier action priority and task +patch omission semantics retain their previous contracts. + +Python retains existing Todo/text codecs and the shared +`actions_are_projection_aligned` owner as named compatibility facts. It does not +select route candidates, deduplicate, count or decide routes. Retire these +compatibility facts only after their shared source owners and final Python +quota assembly callers have native replacements. No provider reads, synthetic +inventory or Goal document scanning are introduced for route-only requests. + +The hint remains advice nested inside the existing quota read model and its +`projection_envelope`, not a new authoritative read API. It grants no execution, +claim, lease or write permission, preserves the durable Goal Next Action and +introduces no persisted-state migration. No leaf RPC or Python fallback owner is +added. Rollback restores the preceding code/artifact and fingerprint; it does +not rewrite Goal state. + +## Costs and qualification + +Initial implementation used `d39662d79784e7afa9dbc844f61452a64cb92c1c`; the +earlier review base was `2c24d5d102dc35c417b3ce49e25096aa37e1642b`. Final +matched CLI evidence uses integration base +`2ce6a5dcd8a5396f5d60312f37958e1b05d529ca` and candidate +`db7f364385bb4418ea223623f0d1ffbd9745a1bd`. + +| Local planning/route segment | Before | Candidate design | +| --- | --- | --- | +| Planning already required | One RPC plus Python route rules | One existing batch RPC | +| Route only | No planning RPC plus Python route rules | One route-only batch RPC | +| Neither required | No RPC | No RPC | + +This is an ownership consolidation, not a claimed latency improvement. Measure +both execution and route-only wait/read branches. Keep the existing warm +non-persistent p95 target below 2 ms and complete-CLI owner-review gates for +p95 regression over 5% or unexplained added 25 ms. Preserve wire, coverage, +fixture and performance budgets; use the complete-snapshot transport rather +than truncating source rows. Do not retain a production dual-owner fallback if +performance fails: optimize the batch or hold/revert the cutover. + +### Verified results and remaining holds + +- **Passed:** generated-contract freshness, 29 contract/catalog tests, both + native-path canary selections, 214 focused Python tests, and all 12 matched + real CLI parity cases across SQLite/File at the revisions above. +- **Passed, focused semantic evidence:** 1,200 deterministic synthetic + Python-baseline versus codec/TS cases; all three independent mutations were + killed (nonempty-list fallback for executable candidates, current/frontier + priority reversal, and nonempty-list fallback for handoff gates). This is not + additional CLI, backend or performance evidence. +- **Passed:** premerge, with an inherited complexity advisory. The full native + suite recorded 6,105 passed, 30 failed and 77 skipped. All 30 failures passed + individually with isolation or a larger temporary filesystem after + runtime/filesystem failures; this does not recast the original aggregate run + as a clean pass. + +The 20-sample full-command measurements retain these threshold exceedances: + +| Case | Base p95 | Candidate p95 | Change | +| --- | --- | --- | --- | +| SQLite current JSON | 879.85 ms | 975.16 ms | +10.83%, +95.31 ms | +| SQLite current Markdown | 851.66 ms | 877.88 ms | +26.23 ms | +| File durable Markdown | 583.29 ms | 618.80 ms | +6.09%, +35.51 ms | + +The other nine cases improved. A separate instrumented 20-pair SQLite current +JSON cohort measured full-command p95 723.73→694.55 ms and planning-RPC p95 +1.294→1.797 ms. These distinct observations do not cancel the original +threshold exceedances or prove their cause. **Performance qualification remains +held for owner review**, not passed; no speedup is claimed. + +Cold-start qualification, clean wheel/sdist installation, PostgreSQL and Host +coverage remain **not_yet_proven**. Keep full-command p50/p95, logical IPC, +wire/output bytes and persisted-state readback distinct from projection timing. +Record Python rules removed, compatibility code added and net production changes +separately from tests, generated files and docs. Later source changes need fresh +exact-head verification. These results grant neither merge nor deployment +authority. diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.zh-CN.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.zh-CN.md new file mode 100644 index 0000000000..db860c39e2 --- /dev/null +++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.zh-CN.md @@ -0,0 +1,82 @@ +# Goal 路由提示并入既有规划批处理 + +## 边界与兼容性 + +本切片将完整只读 Goal 路由投影从 Python 移入 +`work_items/goal_route_hint.ts`，由既有 +`work_item.action_portfolio.project` 批处理消费。Python 路由 builder、task +override 合并器与生产调用者在同一切片切换并退役。既有 `quota.run_decision` +owner 仍产生 task-orchestration override；本切片消费其结果，不重复实现该策略。 + +内部 `quota_planning_packet_request_v1` 包含各自可空的 planning 和 route +context，结果为 `quota_planning_packet_v1`。coordination 合同 JSON 登记两个身份， +既有生成器负责 Python/TypeScript 常量。原有 v0 的要求与结果形态保持原意； +外部嵌套的 `goal_route_hint_v0` 不变。 + +Python 在原 builder 位置、Agent lane、frontier 与 recommendation 选择之后捕获 +route facts。既有 planning 调用在 monitor-only、未准入选择、结算及 cadence 清理 +之前投影 hint；scheduler/序列化阶段不再回填。来源列表缺失与权威空列表保持区别； +空 frontier 对象仍命中 frontier 分支。候选顺序、先选择后截断展示、局部来源计数、 +current action 优先级、canonical frontier action 优先级及 task patch 的省略语义 +保留既有合同。 + +Python 保留既有 Todo/文本 codec，并将共享 `actions_are_projection_aligned` +owner 的结果作为命名兼容事实传递，不选择路由候选、不去重、不计数、不决定路由。 +只有共享来源 owner 与最后的 Python quota 组装调用者均有原生替代后，才能退役这些 +兼容事实。route-only 请求不引入 provider 读取、虚构 inventory 或 Goal 文档扫描。 + +hint 仍是既有 quota 读模型及其 `projection_envelope` 内的建议，不是新增权威读 API。 +它不授予执行、claim、lease 或写权限，保留持久 Goal Next Action，不引入持久化格式 +迁移。不新增 leaf RPC 或 Python 回退 owner。回滚恢复之前的代码/发行物与 fingerprint， +不改写 Goal 状态。 + +## 成本与资格 + +初始实现基线为 `d39662d79784e7afa9dbc844f61452a64cb92c1c`；早期审阅基线为 +`2c24d5d102dc35c417b3ce49e25096aa37e1642b`。最终 CLI 配对证据使用集成基线 +`2ce6a5dcd8a5396f5d60312f37958e1b05d529ca` 与候选 +`db7f364385bb4418ea223623f0d1ffbd9745a1bd`。 + +| 局部 planning/route 段 | 修改前 | 候选设计 | +| --- | --- | --- | +| 已需要 planning | 一次 RPC 加 Python 路由规则 | 一次既有 batch RPC | +| 仅需 route | 零次 planning RPC 加 Python 路由规则 | 一次 route-only batch RPC | +| 两者都不需要 | 零次 RPC | 零次 RPC | + +这是归属收敛，不宣称降低延迟。执行分支与 route-only 等待/只读分支须分别测量。 +保留 warm 非持久化转换 p95 小于 2 ms 的既有目标；完整 CLI p95 回退超过 5% 或 +存在无法解释的 25 ms 增量时，进入 owner review。不得放宽 wire、coverage、fixture +或性能预算；使用完整快照传输，不截断来源行。性能失败不得保留生产双 owner 回退： +应优化批处理，或暂停/回滚切换。 + +### 已验证结果与剩余门禁 + +- **Passed：**生成合同新鲜度、29 项合同/catalog 测试、两个原生路径 canary + 选择、214 项聚焦 Python 测试，以及上述 revision 的 SQLite/File 共 12 项 + 真实 CLI 配对一致性用例全部通过。 +- **Passed，聚焦语义证据：**1,200 项确定性合成 Python baseline 与 codec/TS + 对照通过；三个独立 mutation 均被测试击败（可执行候选按列表非空而非存在回退、 + current/frontier 优先级颠倒、handoff gate 按列表非空而非存在回退）。这些不构成 + 额外的 CLI、后端或性能证据。 +- **Passed：**premerge 通过，保留继承的复杂度 advisory。完整原生测试运行记录为 + 6,105 passed、30 failed、77 skipped。运行时/临时文件系统失败后，30 项失败用例 + 均在隔离或扩大临时文件系统后单独通过；这不将原始整套运行改记为全部通过。 + +20 样本完整命令测量保留以下超阈值结果： + +| 场景 | 基线 p95 | 候选 p95 | 变化 | +| --- | --- | --- | --- | +| SQLite current JSON | 879.85 ms | 975.16 ms | +10.83%，+95.31 ms | +| SQLite current Markdown | 851.66 ms | 877.88 ms | +26.23 ms | +| File durable Markdown | 583.29 ms | 618.80 ms | +6.09%，+35.51 ms | + +其余九项场景改善。另一组带 instrumentation 的 20 对 SQLite current JSON 测量中， +完整命令 p95 为 723.73→694.55 ms，planning RPC p95 为 1.294→1.797 ms。 +这些独立观察不能抵消原始超阈值结果，也不能证明其原因。**性能资格仍待 owner review， +不是 passed**，不宣称性能提升。 + +冷启动资格、干净 wheel/sdist 安装、PostgreSQL 与 Host 覆盖仍为 +**not_yet_proven**。完整命令 p50/p95、逻辑 IPC、wire/输出字节及持久状态 readback +须与投影耗时分开记录。删除的 Python 规则、新增兼容代码、生产代码净变化与测试、 +生成物及文档分别统计。后续源码变更须重新执行 exact-head 验证。这些结果不授予 +合并或部署权限。 diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index a51674dc62..1c006beaaa 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1765,10 +1765,13 @@ The task-scoped `goal_route_hint` override now comes back in that same `quota.run_decision` response. Python projects the lane count from the existing task-orchestration source fact. TypeScript owns the ready-state check, route decision and coordination wording. Python's duplicate `task_goal_route_hint` decision is -deleted, leaving only the bounded merge into the assembled hint. This adds no -request, provider read or authority. Other Goal-route and primary-action -builders remain separate migration work; this does not complete the quota read -transaction or claim a performance gain. +deleted. The complete Goal-route projection and override merge now move to the +existing planning batch, with a route-only internal v1 request when planning +views are absent. This can add a planning RPC on route-only paths; it adds no +provider read or authority. Primary-action builders remain separate work. See +the [Goal-route checkpoint](ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.md) +for compatibility, qualification and cost gates; neither the complete quota +read transaction nor a performance gain is claimed. **Long-history transport boundary.** Replan history still has one TS decision owner. Small requests retain the inline codec; larger complete fact snapshots 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 30eaee9601..f0e5780b21 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 @@ -1263,9 +1263,12 @@ TS 生产者验证器通过共享 guard fixture 执行真实 TS 分类器，并 Task-scoped `goal_route_hint` override 现在由同一个 `quota.run_decision` 响应返回。Python 从既有 task-orchestration source fact 投影 lane 数。TypeScript 持有 ready 状态检查、route 判定和协调说明。Python 重复实现的 -`task_goal_route_hint` 判定已删除，只保留将有界补丁合并进已组装 hint 的步骤。 -本改动不增加请求、provider 读取或 authority。其他 Goal route 与 primary-action -builder 仍是独立迁移工作；这不代表 quota read transaction 已完成，也不宣称性能提升。 +`task_goal_route_hint` 判定已删除。完整 Goal route 投影与 override 合并现迁入既有 +planning batch；没有 planning view 时使用 route-only 内部 v1 请求。这可能为 +route-only 路径增加一次 planning RPC，但不增加 provider 读取或 authority。 +primary-action builder 仍是独立工作。兼容性、资格及成本门见 +[Goal route 检查点](ledger/typescript-control-plane-migration-v0/2026-11-05-goal-route-planning-batch.zh-CN.md)； +不宣称完整 quota read transaction 已完成或性能提升。 **长历史传输边界。** Replan 历史仍由一个 TS owner 决策。小请求保留 inline codec；较大的完整事实快照通过私有临时文件和摘要绑定的引用传递。同一 reducer diff --git a/examples/control_plane/peer-agent-continuation-state-machine-smoke.py b/examples/control_plane/peer-agent-continuation-state-machine-smoke.py index c347d273c3..ea8f6be232 100644 --- a/examples/control_plane/peer-agent-continuation-state-machine-smoke.py +++ b/examples/control_plane/peer-agent-continuation-state-machine-smoke.py @@ -3,6 +3,10 @@ from __future__ import annotations +import hashlib +import json +import shlex +import subprocess import sys import tempfile from pathlib import Path @@ -86,7 +90,7 @@ def write_fixture(root: Path, *, include_review: bool = False) -> tuple[Path, Pa def should_run(project: Path, runtime: Path, registry_path: Path, agent_id: str) -> dict: - return run_cli( + payload = run_cli( "quota", "should-run", "--goal-id", @@ -98,6 +102,39 @@ def should_run(project: Path, runtime: Path, registry_path: Path, agent_id: str) runtime=runtime, cwd=project, ) + if payload.get("schema_version") != "context_capture_navigation_v1": + return payload + # Read the original capture, never rerun admission to recover its decision. + # Full traversal here is a semantic test oracle, not a model read requirement. + command = payload["recovery"]["command"] + content = bytearray() + identity = None + while True: + words = shlex.split(command) + quota_index = words.index("quota") + assert words[quota_index + 1] == "context-read", command + result = subprocess.run( + words, cwd=REPO_ROOT, capture_output=True, text=True, + encoding="utf-8", errors="strict", check=False, timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr + page = json.loads(result.stdout) + assert page["ok"] is True and page["offset"] == len(content), page + page_identity = (page["sha256"], page["total_bytes"]) + if identity is None: + identity = page_identity + assert page_identity == identity, page + chunk = page["content"].encode("utf-8") + assert chunk, "context read made no progress" + content.extend(chunk) + if page["eof"] is True: + assert len(content) == identity[1], page + assert hashlib.sha256(content).hexdigest() == identity[0], page + resolved = json.loads(content) + assert isinstance(resolved, dict), resolved + return resolved + assert page["next_offset"] == len(content) < identity[1], page + command = page["next_command"] def assert_same_peer_continuation() -> None: diff --git a/loopx/canary/planner.py b/loopx/canary/planner.py index d2ac7ed152..9c294a049b 100644 --- a/loopx/canary/planner.py +++ b/loopx/canary/planner.py @@ -310,7 +310,7 @@ "loopx/control_plane/work_items/interaction_contract.py", "loopx/control_plane/todos/handoff_gate.py", "loopx/control_plane/work_items/execution_obligation.py", - "loopx/control_plane/work_items/goal_route_hint.py", + "loopx/control_plane/work_items/goal_route_hint.ts", "loopx/control_plane/work_items/delivery_history.py", "loopx/control_plane/work_items/delivery_history.ts", "loopx/control_plane/work_items/work_lane.py", diff --git a/loopx/canary/qualification_profiles.py b/loopx/canary/qualification_profiles.py index ef5c288b6b..c6e04b2e50 100644 --- a/loopx/canary/qualification_profiles.py +++ b/loopx/canary/qualification_profiles.py @@ -105,7 +105,7 @@ "loopx/control_plane/testing/decision_replay.py", "loopx/control_plane/work_items/interaction_contract.py", "loopx/control_plane/work_items/execution_obligation.py", - "loopx/control_plane/work_items/goal_route_hint.py", + "loopx/control_plane/work_items/goal_route_hint.ts", "loopx/control_plane/work_items/delivery_history.py", "loopx/control_plane/work_items/delivery_history.ts", "loopx/control_plane/work_items/work_lane.py", diff --git a/loopx/control_plane/coordination/coordination_state_contract.generated.ts b/loopx/control_plane/coordination/coordination_state_contract.generated.ts index 79c81a5ee8..e011f7fa5e 100644 --- a/loopx/control_plane/coordination/coordination_state_contract.generated.ts +++ b/loopx/control_plane/coordination/coordination_state_contract.generated.ts @@ -110,6 +110,8 @@ export const ACTION_PORTFOLIO_SELECTION_REQUEST_SCHEMA = "action_selection_quali export const ACTION_PORTFOLIO_SELECTION_RESULT_SCHEMA = "action_selection_qualification_v0"; export const ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_SCHEMA = "quota_planning_packet_request_v0"; export const ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_SCHEMA = "quota_planning_packet_v0"; +export const ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_V1_SCHEMA = "quota_planning_packet_request_v1"; +export const ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_V1_SCHEMA = "quota_planning_packet_v1"; export const TODO_RESUME_NORMALIZE_REQUEST_SCHEMA = "todo_resume_normalize_request_v0"; export const TODO_RESUME_EVALUATION_REQUEST_SCHEMA = "todo_resume_evaluation_request_v0"; @@ -355,7 +357,9 @@ export const COORDINATION_STATE_CONTRACT = deepFreeze({ "selection_request_schema": ACTION_PORTFOLIO_SELECTION_REQUEST_SCHEMA, "selection_result_schema": ACTION_PORTFOLIO_SELECTION_RESULT_SCHEMA, "planning_packet_request_schema": ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_SCHEMA, - "planning_packet_result_schema": ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_SCHEMA + "planning_packet_result_schema": ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_SCHEMA, + "planning_packet_request_v1_schema": ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_V1_SCHEMA, + "planning_packet_result_v1_schema": ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_V1_SCHEMA }, "todo_resume_protocol": { "normalize_request_schema": TODO_RESUME_NORMALIZE_REQUEST_SCHEMA, diff --git a/loopx/control_plane/coordination/coordination_state_contract_generated.py b/loopx/control_plane/coordination/coordination_state_contract_generated.py index 8fb413c7c8..0974bdc43e 100644 --- a/loopx/control_plane/coordination/coordination_state_contract_generated.py +++ b/loopx/control_plane/coordination/coordination_state_contract_generated.py @@ -198,7 +198,9 @@ def _freeze(value: Any) -> Any: 'action_portfolio_protocol': {'selection_request_schema': 'action_selection_qualification_request_v0', 'selection_result_schema': 'action_selection_qualification_v0', 'planning_packet_request_schema': 'quota_planning_packet_request_v0', - 'planning_packet_result_schema': 'quota_planning_packet_v0'}, + 'planning_packet_result_schema': 'quota_planning_packet_v0', + 'planning_packet_request_v1_schema': 'quota_planning_packet_request_v1', + 'planning_packet_result_v1_schema': 'quota_planning_packet_v1'}, 'todo_resume_protocol': {'normalize_request_schema': 'todo_resume_normalize_request_v0', 'evaluation_request_schema': 'todo_resume_evaluation_request_v0', 'evaluation_result_schema': 'todo_resume_evaluation_v0', @@ -316,6 +318,8 @@ def _freeze(value: Any) -> Any: ACTION_PORTFOLIO_SELECTION_RESULT_SCHEMA: Final[str] = 'action_selection_qualification_v0' ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_SCHEMA: Final[str] = 'quota_planning_packet_request_v0' ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_SCHEMA: Final[str] = 'quota_planning_packet_v0' +ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_V1_SCHEMA: Final[str] = 'quota_planning_packet_request_v1' +ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_V1_SCHEMA: Final[str] = 'quota_planning_packet_v1' TODO_RESUME_NORMALIZE_REQUEST_SCHEMA: Final[str] = 'todo_resume_normalize_request_v0' TODO_RESUME_EVALUATION_REQUEST_SCHEMA: Final[str] = 'todo_resume_evaluation_request_v0' diff --git a/loopx/control_plane/coordination/coordination_state_contract_v0.json b/loopx/control_plane/coordination/coordination_state_contract_v0.json index e8fde13ac4..01f4abe5a7 100644 --- a/loopx/control_plane/coordination/coordination_state_contract_v0.json +++ b/loopx/control_plane/coordination/coordination_state_contract_v0.json @@ -220,7 +220,9 @@ "selection_request_schema": "action_selection_qualification_request_v0", "selection_result_schema": "action_selection_qualification_v0", "planning_packet_request_schema": "quota_planning_packet_request_v0", - "planning_packet_result_schema": "quota_planning_packet_v0" + "planning_packet_result_schema": "quota_planning_packet_v0", + "planning_packet_request_v1_schema": "quota_planning_packet_request_v1", + "planning_packet_result_v1_schema": "quota_planning_packet_v1" }, "todo_resume_protocol": { "normalize_request_schema": "todo_resume_normalize_request_v0", diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index f90a72d0c1..ea90cd8e1c 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -77,7 +77,6 @@ ) from ..quota.task_orchestration import ( PEER_COORDINATION_BLOCKED_ACTION, - apply_task_route_hint_override, attach_task_orchestration_payload, task_orchestration_contract_is_actionable, task_orchestration_requires_material_change_stop, @@ -118,6 +117,7 @@ from ..turn_driver.delivery_continuity import evaluate_delivery_route from ..work_items.action_portfolio import ( build_quota_planning_packet, + capture_goal_route_context, qualify_action_selection_from_inventory, ) from ..work_items.execution_obligation import build_execution_obligation @@ -125,7 +125,6 @@ action_selection_context, apply_action_selection_recovery_projection, ) -from ..work_items.goal_route_hint import build_goal_route_hint from ..work_items.interaction_contract import ( _build_interaction_contract_projection, unadmitted_action_selection, @@ -803,7 +802,7 @@ class _QuotaDecisionRoute: active_state_next_action_text: str | None latest_run_recommended_action_text: str | None next_action_warning: dict[str, Any] | None - goal_route_hint: dict[str, Any] | None + goal_route_context: dict[str, Any] | None payload_work_lane_contract: dict[str, Any] | None @@ -832,6 +831,7 @@ def _planning_projections( ) return build_quota_planning_packet( projection_enabled=projection_enabled, + goal_route_context=route.goal_route_context, include_detail=include_detail, goal_id=prepared.safe_goal_id, selected=route.agent_lane_next_action, @@ -961,7 +961,7 @@ def _resolve_quota_should_run_route( item.get("latest_run_recommended_action") or prepared.project_asset.get("latest_run_recommended_action"), limit=320, ), - next_action_warning=None, goal_route_hint=None, + next_action_warning=None, goal_route_context=None, payload_work_lane_contract=_payload_work_lane_contract( prepared.work_lane_contract, effective_action=fields["effective_action"], recovery_allowed=False, agent_scope_frontier=None, @@ -1274,7 +1274,7 @@ def _resolve_quota_should_run_route( latest_run_recommended_action=latest_run_recommended_action_text, agent_lane_next_action=agent_lane_next_action, ) - goal_route_hint = build_goal_route_hint( + goal_route_context = capture_goal_route_context( agent_identity=prepared.agent_identity, agent_todo_summary=prepared.agent_todo_summary, agent_lane_next_action=agent_lane_next_action, @@ -1283,10 +1283,7 @@ def _resolve_quota_should_run_route( active_state_next_action=active_state_next_action_text, latest_run_recommended_action=latest_run_recommended_action_text, selected_recommended_action=selected_recommended_action, - ) - goal_route_hint = apply_task_route_hint_override( - goal_route_hint, - run_decision.goal_route_hint_override, + goal_route_hint_override=run_decision.goal_route_hint_override, ) payload_work_lane_contract = _payload_work_lane_contract( prepared.work_lane_contract, @@ -1318,7 +1315,7 @@ def _resolve_quota_should_run_route( active_state_next_action_text=active_state_next_action_text, latest_run_recommended_action_text=latest_run_recommended_action_text, next_action_warning=next_action_warning, - goal_route_hint=goal_route_hint, + goal_route_context=goal_route_context, payload_work_lane_contract=payload_work_lane_contract, ) @@ -1529,7 +1526,6 @@ def _build_active_quota_payload( _attach_truthy_fields( payload, agent_lane_frontier_hint=route.agent_lane_frontier_hint, - goal_route_hint=route.goal_route_hint, agent_scope_frontier=route.agent_scope_frontier, workspace_guard=prepared.workspace_guard, automation_prompt_upgrade=prepared.automation_prompt_upgrade, diff --git a/loopx/control_plane/quota/task_orchestration.py b/loopx/control_plane/quota/task_orchestration.py index 551083a4e3..d3e46c98a7 100644 --- a/loopx/control_plane/quota/task_orchestration.py +++ b/loopx/control_plane/quota/task_orchestration.py @@ -35,23 +35,6 @@ def task_orchestration_contract_is_actionable( return str(contract.get("execution_state") or "ready") == "ready" -def apply_task_route_hint_override( - hint: dict[str, Any] | None, - override: dict[str, Any] | None, -) -> dict[str, Any] | None: - """Apply the route patch returned by the TypeScript decision owner.""" - if hint is None or override is None: - return hint - projected_hint = { - key: value for key, value in hint.items() - if not (override.get("omit_current_agent_next_action") is True - and key == "current_agent_next_action") - } - projected_hint.update({key: value for key, value in override.items() - if key != "omit_current_agent_next_action"}) - return projected_hint - - def task_orchestration_requires_material_change_stop( contract: dict[str, Any] | None, *, diff --git a/loopx/control_plane/work_items/action_portfolio.py b/loopx/control_plane/work_items/action_portfolio.py index 81b367ec75..6e487bf885 100644 --- a/loopx/control_plane/work_items/action_portfolio.py +++ b/loopx/control_plane/work_items/action_portfolio.py @@ -1,17 +1,23 @@ from __future__ import annotations from collections.abc import Mapping +from copy import deepcopy from typing import Any +from ...state_projection import actions_are_projection_aligned +from ..runtime.public_safety import compact_text from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result from ..goals.goal_vision_policy import DEFAULT_PLANNING_HORIZON_GOAL from ..coordination.coordination_state_contract_generated import ( - ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_SCHEMA, - ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_SCHEMA, + ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_V1_SCHEMA, + ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_V1_SCHEMA, ACTION_PORTFOLIO_SELECTION_REQUEST_SCHEMA, ACTION_PORTFOLIO_SELECTION_RESULT_SCHEMA, ) -from ..todos.contract import normalize_todo_id, normalize_todo_task_class +from ..todos.contract import ( + normalize_todo_id, normalize_todo_task_class, normalize_todo_status, + normalize_todo_claimed_by, normalize_todo_blocks_agent, normalize_todo_excluded_agents, +) from .planning_inventory import ( build_quota_planning_inventory_request, compact_planning_candidate, @@ -19,8 +25,8 @@ ACTION_SELECTION_QUALIFICATION_REQUEST_SCHEMA_VERSION = ACTION_PORTFOLIO_SELECTION_REQUEST_SCHEMA ACTION_SELECTION_QUALIFICATION_SCHEMA_VERSION = ACTION_PORTFOLIO_SELECTION_RESULT_SCHEMA -QUOTA_PLANNING_PACKET_REQUEST_SCHEMA_VERSION = ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_SCHEMA -QUOTA_PLANNING_PACKET_SCHEMA_VERSION = ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_SCHEMA +QUOTA_PLANNING_PACKET_REQUEST_SCHEMA_VERSION = ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_V1_SCHEMA +QUOTA_PLANNING_PACKET_SCHEMA_VERSION = ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_V1_SCHEMA RETAINED_ACTION_SELECTION_REENTRY_REQUEST_SCHEMA_VERSION = ( "retained_action_selection_reentry_request_v0" ) @@ -42,6 +48,102 @@ def _frontier_acceptance_gaps( return acceptance_gaps if isinstance(acceptance_gaps, list) else [] +def capture_goal_route_context( + *, + agent_identity: dict[str, Any] | None, + agent_todo_summary: dict[str, Any] | None, + agent_lane_next_action: dict[str, Any] | None, + agent_scope_frontier: dict[str, Any] | None, + agent_lane_frontier_hint: dict[str, Any] | None, + active_state_next_action: Any, + latest_run_recommended_action: Any, + selected_recommended_action: Any, + goal_route_hint_override: dict[str, Any] | None, +) -> dict[str, Any] | None: + """Snapshot compatibility facts, without selecting, counting or routing work. + + Python normalization and text alignment remain shared owners. Every source + retains its order and list-presence semantics; TypeScript owns all choices. + """ + agent_id = normalize_todo_claimed_by( + agent_identity.get("agent_id") if isinstance(agent_identity, dict) else None + ) + if not agent_id: + return None + + def row(value: Any) -> dict[str, Any] | None: + if not isinstance(value, dict): + return None + text = compact_text(str(value.get("text") or ""), limit=220) + return { + "done": value.get("done") is True, + "status": normalize_todo_status(value.get("status")), + "task_class": normalize_todo_task_class( + value.get("task_class"), text=str(value.get("text") or ""), + action_kind=value.get("action_kind"), + ), + "todo_id": normalize_todo_id(value.get("todo_id")), + "claimed_by": normalize_todo_claimed_by(value.get("claimed_by")), + "blocks_agent": normalize_todo_blocks_agent(value.get("blocks_agent")), + "unblocks_todo_id": normalize_todo_id(value.get("unblocks_todo_id")), + "excluded_agents": normalize_todo_excluded_agents(value.get("excluded_agents")), + "text": text, + "display_text": compact_text(text, limit=120), + "gate_identity": str(value.get("todo_id") or value.get("index") or value.get("text") or ""), + "gate_state": value.get("gate_state"), + **{key: value[key] for key in ("priority", "action_kind") if value.get(key) is not None}, + } + + def integer(value: Any) -> int: + try: + return max(0, int(value or 0)) + except (TypeError, ValueError): + return 0 + + summary = agent_todo_summary if isinstance(agent_todo_summary, dict) else {} + summary_facts: dict[str, Any] = {} + for field in ( + "claimed_advancement_open_items", "claimed_by_others_items", + "executable_backlog_items", "unclaimed_priority_open_items", + "current_agent_handoff_gates", "handoff_gates", + ): + values = summary.get(field) + summary_facts[field] = [row(value) for value in values] if isinstance(values, list) else None + for field in ("current_agent_claimed_advancement_count", "claimed_advancement_open_count"): + summary_facts[field] = integer(summary.get(field)) + current = None + if isinstance(agent_lane_next_action, dict): + current = { + "todo_id": normalize_todo_id(agent_lane_next_action.get("todo_id")), + "text": compact_text(str(agent_lane_next_action.get("text") or ""), limit=120), + "claim_required_before_work": agent_lane_next_action.get("claim_required_before_work") is True, + **{key: agent_lane_next_action[key] for key in ("selected_by", "confidence", "source") + if agent_lane_next_action.get(key) is not None}, + } + active = compact_text(str(active_state_next_action or ""), limit=260) + selected = compact_text(str(selected_recommended_action or ""), limit=260) + return deepcopy({ + "agent_id": agent_id, + "agent_todo_summary": summary_facts, + "agent_lane_next_action": current, + "agent_scope_frontier": ( + {key: str(agent_scope_frontier.get(key) or "") for key in ("action", "effective_action", "reason")} + if isinstance(agent_scope_frontier, dict) else None + ), + "agent_lane_frontier_hint": ( + {key: agent_lane_frontier_hint[key] for key in + ("decision", "source", "reason_code", "target_todo_id", "quiet_noop_allowed") + if agent_lane_frontier_hint.get(key) is not None} + if isinstance(agent_lane_frontier_hint, dict) else None + ), + "active_state_next_action": active, + "selected_recommended_action": selected, + "latest_run_recommended_action": compact_text(str(latest_run_recommended_action or ""), limit=260), + "actions_projection_aligned": actions_are_projection_aligned(active, selected), + "goal_route_hint_override": goal_route_hint_override, + }) + + def build_quota_planning_packet( *, projection_enabled: bool, @@ -57,44 +159,48 @@ def build_quota_planning_packet( replan_selection_context: Mapping[str, Any] | None = None, source_status: Mapping[str, Any] | None = None, planning_horizon_goal: bool = DEFAULT_PLANNING_HORIZON_GOAL, + goal_route_context: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Project every requested planning lens through one TypeScript request.""" if not isinstance(planning_horizon_goal, bool): raise ValueError("planning_horizon_goal must be a boolean") - if not projection_enabled and not include_detail and replan_selection_context is None: - return {} - inventory_request = build_quota_planning_inventory_request( - goal_id=goal_id, - selected=selected, - agent_id=agent_id, - agent_todo_summary=agent_todo_summary, - agent_todo_source_items=agent_todo_source_items, - capability_gate=capability_gate, - blocked_priority_fallback=blocked_priority_fallback, + planning_requested = projection_enabled or include_detail or replan_selection_context is not None + inventory_request = ( + build_quota_planning_inventory_request( + goal_id=goal_id, selected=selected, agent_id=agent_id, + agent_todo_summary=agent_todo_summary, + agent_todo_source_items=agent_todo_source_items, + capability_gate=capability_gate, + blocked_priority_fallback=blocked_priority_fallback, + ) if planning_requested else None ) - if inventory_request is None: - return {} - from .replan_context_codec import goal_context_facts + planning = None + if inventory_request is not None: + from .replan_context_codec import goal_context_facts - goal_facts = ( - goal_context_facts(source_status, goal_id, (agent_todo_summary or {}).get("goal_acceptance_contract")) - if planning_horizon_goal and source_status and source_status.get("registry") else None - ) + goal_facts = ( + goal_context_facts(source_status, goal_id, (agent_todo_summary or {}).get("goal_acceptance_contract")) + if planning_horizon_goal and source_status and source_status.get("registry") else None + ) + planning = { + "planning_inventory_request": inventory_request, + "projection_enabled": projection_enabled, + "include_detail": include_detail, + "acceptance_gaps": _frontier_acceptance_gaps(goal_frontier_projection), + **({"goal_facts": goal_facts} if goal_facts is not None else {}), + **({"replan_selection_context": dict(replan_selection_context)} + if replan_selection_context is not None else {}), + } + if planning is None and goal_route_context is None: + return {} try: projected = effect_runtime_result( "work_item.action_portfolio.project", { "schema_version": QUOTA_PLANNING_PACKET_REQUEST_SCHEMA_VERSION, - "planning_inventory_request": inventory_request, - "projection_enabled": projection_enabled, - "include_detail": include_detail, - "acceptance_gaps": _frontier_acceptance_gaps( - goal_frontier_projection - ), - **({"goal_facts": goal_facts} if goal_facts is not None else {}), - **({"replan_selection_context": dict(replan_selection_context)} - if replan_selection_context is not None else {}), + "planning": planning, + "goal_route_context": dict(goal_route_context) if goal_route_context is not None else None, }, ) except EffectRuntimeRejected as exc: @@ -105,6 +211,7 @@ def build_quota_planning_packet( raise RuntimeError("TypeScript quota planning packet shape mismatch") result: dict[str, Any] = {} for field in ( + "goal_route_hint", "action_portfolio", "planning_horizon", "agent_todo_planning_inventory", diff --git a/loopx/control_plane/work_items/action_portfolio.ts b/loopx/control_plane/work_items/action_portfolio.ts index 1d56a9ee57..06257b81e0 100644 --- a/loopx/control_plane/work_items/action_portfolio.ts +++ b/loopx/control_plane/work_items/action_portfolio.ts @@ -11,6 +11,8 @@ import { import type { JsonObject } from "../effect_program.ts"; import { ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_SCHEMA, + ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_V1_SCHEMA, + ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_V1_SCHEMA, ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_SCHEMA, ACTION_PORTFOLIO_SELECTION_REQUEST_SCHEMA, ACTION_PORTFOLIO_SELECTION_RESULT_SCHEMA, @@ -24,6 +26,7 @@ import { PLANNING_HORIZON_REQUEST_SCHEMA_VERSION, projectQuotaPlanningHorizon, } from "./planning_horizon.ts"; +import { projectGoalRouteHint } from "./goal_route_hint.ts"; import { EffectiveAction } from "../quota/effective_action.generated.ts"; export const ACTION_PORTFOLIO_SCHEMA_VERSION = "quota_action_portfolio_v2"; @@ -341,6 +344,20 @@ function projectQuotaPlanningPacket(request: JsonObject): JsonObject { */ export function projectQuotaActionPortfolio(value: unknown): JsonObject | null { const request = requireJsonObject(value, "action_portfolio_request"); + if (request.schema_version === ACTION_PORTFOLIO_PLANNING_PACKET_REQUEST_V1_SCHEMA) { + // Decode both optional blocks at the boundary. A route-only packet must not + // synthesize an inventory or run any planning/Goal-context projection. + const planning = request.planning === null ? null + : requireJsonObject(request.planning, "planning_packet_request.planning"); + const route = request.goal_route_context === null ? null + : projectGoalRouteHint(request.goal_route_context); + const projected = planning === null ? {} : projectQuotaPlanningPacket(planning); + return { + ...projected, + schema_version: ACTION_PORTFOLIO_PLANNING_PACKET_RESULT_V1_SCHEMA, + ...(route === null ? {} : { goal_route_hint: route }), + }; + } if (request.schema_version === QUOTA_PLANNING_PACKET_REQUEST_SCHEMA_VERSION) { return projectQuotaPlanningPacket(request); } diff --git a/loopx/control_plane/work_items/goal_route_hint.py b/loopx/control_plane/work_items/goal_route_hint.py deleted file mode 100644 index 4080bdd551..0000000000 --- a/loopx/control_plane/work_items/goal_route_hint.py +++ /dev/null @@ -1,317 +0,0 @@ -from __future__ import annotations - -from typing import Any - -from ...state_projection import actions_are_projection_aligned -from ..agents.agent_scope_frontier import read_frontier_action -from ..todos.contract import ( - TODO_STATUS_OPEN, - TODO_TASK_CLASS_ADVANCEMENT, - normalize_todo_blocks_agent, - normalize_todo_claimed_by, - normalize_todo_excluded_agents, - normalize_todo_id, - normalize_todo_status, - normalize_todo_task_class, -) -from ..todos.handoff_gate import HandoffGateState - - -GOAL_ROUTE_HINT_SCHEMA_VERSION = "goal_route_hint_v0" -GOAL_ROUTE_HINT_ITEM_LIMIT = 3 - - -def _compact_text(value: Any, *, limit: int = 220) -> str: - text = " ".join(str(value or "").strip().split()) - if len(text) <= limit: - return text - return text[: limit - 1].rstrip() + "…" - - -def _int_count(value: Any) -> int: - try: - return max(0, int(value or 0)) - except (TypeError, ValueError): - return 0 - - -def _is_actionable_advancement(item: dict[str, Any]) -> bool: - if not isinstance(item, dict): - return False - if item.get("done") is True: - return False - status = normalize_todo_status(item.get("status")) - if status and status != TODO_STATUS_OPEN: - return False - return ( - normalize_todo_task_class( - item.get("task_class"), - text=str(item.get("text") or ""), - action_kind=item.get("action_kind"), - ) - == TODO_TASK_CLASS_ADVANCEMENT - ) - - -def _compact_todo(item: dict[str, Any]) -> dict[str, Any] | None: - if not _is_actionable_advancement(item): - return None - text = _compact_text(item.get("text"), limit=220) - if not text: - return None - payload: dict[str, Any] = {} - for key, normalizer in ( - ("todo_id", normalize_todo_id), - ("claimed_by", normalize_todo_claimed_by), - ("blocks_agent", normalize_todo_blocks_agent), - ("unblocks_todo_id", normalize_todo_id), - ): - normalized = normalizer(item.get(key)) - if normalized: - payload[key] = normalized - excluded_agents = normalize_todo_excluded_agents(item.get("excluded_agents")) - if excluded_agents: - payload["excluded_agents"] = excluded_agents - for key in ("priority", "action_kind"): - if item.get(key) is not None: - payload[key] = item.get(key) - if not payload.get("todo_id"): - payload["text"] = _compact_text(text, limit=120) - return payload - - -def _todo_items( - items: Any, - *, - agent_id: str, - claim: str, - limit: int = GOAL_ROUTE_HINT_ITEM_LIMIT, -) -> list[dict[str, Any]]: - if not isinstance(items, list): - return [] - selected: list[dict[str, Any]] = [] - seen: set[str] = set() - for item in items: - if not isinstance(item, dict): - continue - claimed_by = normalize_todo_claimed_by(item.get("claimed_by")) - if claim == "current" and claimed_by != agent_id: - continue - if claim == "other" and (not claimed_by or claimed_by == agent_id): - continue - if claim == "unclaimed" and claimed_by: - continue - compact = _compact_todo(item) - if not compact: - continue - identity = str(compact.get("todo_id") or compact.get("text") or "") - if identity in seen: - continue - seen.add(identity) - selected.append(compact) - if len(selected) >= limit: - break - return selected - - -def _count_advancement_items(items: Any, *, agent_id: str | None = None, claim: str | None = None) -> int: - if not isinstance(items, list): - return 0 - count = 0 - for item in items: - if not isinstance(item, dict) or not _is_actionable_advancement(item): - continue - claimed_by = normalize_todo_claimed_by(item.get("claimed_by")) - if claim == "unclaimed" and claimed_by: - continue - if claim == "other" and (not claimed_by or claimed_by == agent_id): - continue - if claim == "current" and claimed_by != agent_id: - continue - count += 1 - return count - - -def _current_action(agent_lane_next_action: dict[str, Any] | None) -> dict[str, Any] | None: - if not isinstance(agent_lane_next_action, dict): - return None - action: dict[str, Any] = {} - todo_id = normalize_todo_id(agent_lane_next_action.get("todo_id")) - if todo_id: - action["todo_id"] = todo_id - for key in ("selected_by", "confidence", "source"): - if agent_lane_next_action.get(key) is not None: - action[key] = agent_lane_next_action.get(key) - if not action.get("todo_id"): - text = _compact_text(agent_lane_next_action.get("text"), limit=120) - if not text: - return None - action["text"] = text - if agent_lane_next_action.get("claim_required_before_work") is True: - action["claim_required_before_work"] = True - return action - - -def _blocking_handoff_gates(agent_todo_summary: dict[str, Any], *, agent_id: str) -> list[dict[str, Any]]: - gates = agent_todo_summary.get("current_agent_handoff_gates") - if not isinstance(gates, list): - gates = agent_todo_summary.get("handoff_gates") - if not isinstance(gates, list): - return [] - selected: list[dict[str, Any]] = [] - seen: set[str] = set() - for item in gates: - if not isinstance(item, dict): - continue - if agent_id not in normalize_todo_excluded_agents(item.get("excluded_agents")): - continue - if item.get("gate_state") != HandoffGateState.BLOCKING.value: - continue - identity = str(item.get("todo_id") or item.get("index") or item.get("text") or "") - if identity in seen: - continue - seen.add(identity) - compact: dict[str, Any] = {} - for key, normalizer in ( - ("todo_id", normalize_todo_id), - ("claimed_by", normalize_todo_claimed_by), - ("unblocks_todo_id", normalize_todo_id), - ): - normalized = normalizer(item.get(key)) - if normalized: - compact[key] = normalized - excluded_agents = normalize_todo_excluded_agents(item.get("excluded_agents")) - if excluded_agents: - compact["excluded_agents"] = excluded_agents - if item.get("gate_state") is not None: - compact["gate_state"] = item.get("gate_state") - selected.append(compact) - if len(selected) >= GOAL_ROUTE_HINT_ITEM_LIMIT: - break - return selected - - -def build_goal_route_hint( - *, - agent_identity: dict[str, Any] | None, - agent_todo_summary: dict[str, Any] | None, - agent_lane_next_action: dict[str, Any] | None, - agent_scope_frontier: dict[str, Any] | None, - agent_lane_frontier_hint: dict[str, Any] | None, - active_state_next_action: Any, - latest_run_recommended_action: Any, - selected_recommended_action: Any, -) -> dict[str, Any] | None: - if not isinstance(agent_identity, dict): - return None - agent_id = normalize_todo_claimed_by(agent_identity.get("agent_id")) - if not agent_id: - return None - agent_summary = agent_todo_summary if isinstance(agent_todo_summary, dict) else {} - current_action = _current_action(agent_lane_next_action) - other_agent_actions = _todo_items( - agent_summary.get("claimed_advancement_open_items"), - agent_id=agent_id, - claim="other", - ) - if not other_agent_actions: - other_agent_actions = _todo_items( - agent_summary.get("claimed_by_others_items"), - agent_id=agent_id, - claim="other", - ) - executable_items = agent_summary.get("executable_backlog_items") - unclaimed_source = ( - executable_items - if isinstance(executable_items, list) - else agent_summary.get("unclaimed_priority_open_items") - ) - unclaimed_actions = _todo_items( - unclaimed_source, - agent_id=agent_id, - claim="unclaimed", - ) - blocking_handoff_gates = _blocking_handoff_gates(agent_summary, agent_id=agent_id) - current_count = _int_count(agent_summary.get("current_agent_claimed_advancement_count")) - unclaimed_count = _count_advancement_items( - unclaimed_source, - agent_id=agent_id, - claim="unclaimed", - ) - other_count = _int_count(agent_summary.get("claimed_advancement_open_count")) - if not other_count: - other_count = _count_advancement_items( - agent_summary.get("claimed_advancement_open_items"), - agent_id=agent_id, - claim="other", - ) - else: - other_count = max(0, other_count - current_count) - other_count = max(other_count, len(other_agent_actions)) - - route_decision = "durable_goal_route" - reason = "no agent-lane override is needed; preserve durable goal Next Action" - if current_action: - if current_action.get("claim_required_before_work") is True: - route_decision = "claim_then_run_current_agent" - reason = "quota selected an unclaimed in-scope todo for this agent lane" - else: - route_decision = "run_current_agent_lane" - reason = "quota selected a runnable current-agent lane todo" - elif isinstance(agent_scope_frontier, dict): - route_decision = read_frontier_action(agent_scope_frontier) or "agent_scope_frontier" - reason = str(agent_scope_frontier.get("reason") or "agent-scope frontier blocks delivery") - elif blocking_handoff_gates or other_agent_actions: - route_decision = "wait_or_reassign_other_agent_lane" - reason = "visible advancement work is owned by another agent lane" - elif unclaimed_actions: - route_decision = "claim_unowned_in_scope" - reason = "unclaimed advancement work is available to the current agent lane" - - active_text = _compact_text(active_state_next_action, limit=260) - selected_text = _compact_text(selected_recommended_action, limit=260) - latest_text = _compact_text(latest_run_recommended_action, limit=260) - hint: dict[str, Any] = { - "schema_version": GOAL_ROUTE_HINT_SCHEMA_VERSION, - "kind": "agent_lane_synthesis", - "agent_id": agent_id, - "agent_model": "peer_v1", - "route_decision": route_decision, - "reason": reason, - "preserves_goal_next_action": True, - "goal_next_action_mutation": "none", - "counts": { - "current_agent_claimed_advancement_count": current_count, - "unclaimed_advancement_count": unclaimed_count, - "other_agent_claimed_advancement_count": other_count, - "blocking_handoff_gate_count": len(blocking_handoff_gates), - }, - "selected_action_differs_from_durable": bool( - active_text - and selected_text - and not actions_are_projection_aligned(active_text, selected_text) - ), - "has_durable_next_action": bool(active_text), - "has_latest_run_recommended_action": bool(latest_text), - } - if current_action: - hint["current_agent_next_action"] = current_action - if other_agent_actions: - hint["other_agent_next_actions"] = other_agent_actions - if unclaimed_actions: - hint["unclaimed_next_actions"] = unclaimed_actions - if blocking_handoff_gates: - hint["blocking_handoff_gates"] = blocking_handoff_gates - if isinstance(agent_lane_frontier_hint, dict): - hint["frontier_hint"] = { - key: agent_lane_frontier_hint[key] - for key in ( - "decision", - "source", - "reason_code", - "target_todo_id", - "quiet_noop_allowed", - ) - if agent_lane_frontier_hint.get(key) is not None - } - return hint diff --git a/loopx/control_plane/work_items/goal_route_hint.ts b/loopx/control_plane/work_items/goal_route_hint.ts new file mode 100644 index 0000000000..302e9ee04d --- /dev/null +++ b/loopx/control_plane/work_items/goal_route_hint.ts @@ -0,0 +1,272 @@ +/** Read-only Goal routing inside the quota planning batch, never execution authority. */ +import type { JsonObject } from "../effect_program.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { requireBoolean, requireInteger, requireJsonObject, requireStringArray } from "../runtime_decode.ts"; + +export const GOAL_ROUTE_HINT_SCHEMA_VERSION = "goal_route_hint_v0"; +const ITEM_LIMIT = 3; + +interface RouteRow extends JsonObject { + done: boolean; + status: string | null; + task_class: string; + todo_id: string | null; + claimed_by: string | null; + blocks_agent: string | null; + unblocks_todo_id: string | null; + excluded_agents: string[]; + text: string; + display_text: string; + gate_identity: string; + gate_state: unknown; +} +interface CurrentAction extends JsonObject { + todo_id: string | null; + text: string; + claim_required_before_work: boolean; +} +interface Frontier { action: string; effective_action: string; reason: string } +interface TaskRouteOverride extends JsonObject { + omit_current_agent_next_action?: boolean; + kind?: string; + route_decision?: string; + reason?: string; + peer_lane_count?: number; +} +export interface GoalRouteContext { + agent_id: string | null; + agent_todo_summary: { + claimed_advancement_open_items: (RouteRow | null)[] | null; + claimed_by_others_items: (RouteRow | null)[] | null; + executable_backlog_items: (RouteRow | null)[] | null; + unclaimed_priority_open_items: (RouteRow | null)[] | null; + current_agent_handoff_gates: (RouteRow | null)[] | null; + handoff_gates: (RouteRow | null)[] | null; + current_agent_claimed_advancement_count: number; + claimed_advancement_open_count: number; + }; + agent_lane_next_action: CurrentAction | null; + agent_scope_frontier: Frontier | null; + agent_lane_frontier_hint: JsonObject | null; + active_state_next_action: string; + selected_recommended_action: string; + latest_run_recommended_action: string; + actions_projection_aligned: boolean; + goal_route_hint_override: TaskRouteOverride | null; +} +function text(value: unknown, label: string): string { + if (typeof value !== "string") throw new EffectRuntimeRequestError(`${label} must be a string`); + return value; +} +function nullableText(value: unknown, label: string): string | null { + return value === null ? null : text(value, label); +} +function count(value: unknown, label: string): number { + const result = requireInteger(value, label); + if (result < 0) throw new EffectRuntimeRequestError(`${label} must be nonnegative`); + return result; +} +function rows(value: unknown, label: string): (RouteRow | null)[] | null { + if (value === null) return null; + if (!Array.isArray(value)) throw new EffectRuntimeRequestError(`${label} must be an array or null`); + return value.map((entry, index) => { + if (entry === null) return null; + const path = `${label}[${index}]`, raw = requireJsonObject(entry, path); + return { + done: requireBoolean(raw.done, `${path}.done`), + status: nullableText(raw.status, `${path}.status`), + task_class: text(raw.task_class, `${path}.task_class`), + todo_id: nullableText(raw.todo_id, `${path}.todo_id`), + claimed_by: nullableText(raw.claimed_by, `${path}.claimed_by`), + blocks_agent: nullableText(raw.blocks_agent, `${path}.blocks_agent`), + unblocks_todo_id: nullableText(raw.unblocks_todo_id, `${path}.unblocks_todo_id`), + excluded_agents: requireStringArray(raw.excluded_agents, `${path}.excluded_agents`), + text: text(raw.text, `${path}.text`), + display_text: text(raw.display_text, `${path}.display_text`), + gate_identity: text(raw.gate_identity, `${path}.gate_identity`), + gate_state: raw.gate_state, + ...(raw.priority == null ? {} : { priority: raw.priority }), + ...(raw.action_kind == null ? {} : { action_kind: raw.action_kind }), + }; + }); +} +function override(value: unknown): TaskRouteOverride | null { + if (value === null) return null; + const raw = requireJsonObject(value, "goal_route_hint_override"); + const result: TaskRouteOverride = {}; + for (const field of ["kind", "route_decision", "reason"] as const) { + if (raw[field] !== undefined) result[field] = text(raw[field], `goal_route_hint_override.${field}`); + } + if (raw.omit_current_agent_next_action !== undefined) result.omit_current_agent_next_action = + requireBoolean(raw.omit_current_agent_next_action, "goal_route_hint_override.omit_current_agent_next_action"); + if (raw.peer_lane_count !== undefined) result.peer_lane_count = count(raw.peer_lane_count, "goal_route_hint_override.peer_lane_count"); + for (const field of Object.keys(raw)) { + if (!Object.hasOwn(result, field)) throw new EffectRuntimeRequestError(`unsupported goal_route_hint_override.${field}`); + } + return result; +} +export function decodeGoalRouteContext(value: unknown): GoalRouteContext { + const raw = requireJsonObject(value, "goal_route_context"); + const summary = requireJsonObject(raw.agent_todo_summary, "agent_todo_summary"); + let current: CurrentAction | null = null; + if (raw.agent_lane_next_action !== null) { + const item = requireJsonObject(raw.agent_lane_next_action, "agent_lane_next_action"); + current = { + todo_id: nullableText(item.todo_id, "agent_lane_next_action.todo_id"), + text: text(item.text, "agent_lane_next_action.text"), + claim_required_before_work: requireBoolean(item.claim_required_before_work, "agent_lane_next_action.claim_required_before_work"), + }; + for (const field of ["selected_by", "confidence", "source"] as const) { + if (item[field] != null) current[field] = item[field]; + } + } + let frontier: Frontier | null = null; + if (raw.agent_scope_frontier !== null) { + const item = requireJsonObject(raw.agent_scope_frontier, "agent_scope_frontier"); + frontier = { + action: text(item.action ?? "", "agent_scope_frontier.action"), + effective_action: text(item.effective_action ?? "", "agent_scope_frontier.effective_action"), + reason: text(item.reason ?? "", "agent_scope_frontier.reason"), + }; + } + return { + agent_id: nullableText(raw.agent_id, "agent_id"), + agent_todo_summary: { + claimed_advancement_open_items: rows(summary.claimed_advancement_open_items, "claimed_advancement_open_items"), + claimed_by_others_items: rows(summary.claimed_by_others_items, "claimed_by_others_items"), + executable_backlog_items: rows(summary.executable_backlog_items, "executable_backlog_items"), + unclaimed_priority_open_items: rows(summary.unclaimed_priority_open_items, "unclaimed_priority_open_items"), + current_agent_handoff_gates: rows(summary.current_agent_handoff_gates, "current_agent_handoff_gates"), + handoff_gates: rows(summary.handoff_gates, "handoff_gates"), + current_agent_claimed_advancement_count: count(summary.current_agent_claimed_advancement_count, "current_agent_claimed_advancement_count"), + claimed_advancement_open_count: count(summary.claimed_advancement_open_count, "claimed_advancement_open_count"), + }, + agent_lane_next_action: current, + agent_scope_frontier: frontier, + agent_lane_frontier_hint: raw.agent_lane_frontier_hint === null ? null : requireJsonObject(raw.agent_lane_frontier_hint, "agent_lane_frontier_hint"), + active_state_next_action: text(raw.active_state_next_action, "active_state_next_action"), + selected_recommended_action: text(raw.selected_recommended_action, "selected_recommended_action"), + latest_run_recommended_action: text(raw.latest_run_recommended_action, "latest_run_recommended_action"), + actions_projection_aligned: requireBoolean(raw.actions_projection_aligned, "actions_projection_aligned"), + goal_route_hint_override: override(raw.goal_route_hint_override), + }; +} +function actionable(row: RouteRow): boolean { + return !row.done && (!row.status || row.status === "open") && row.task_class === "advancement_task"; +} +function matchesClaim(row: RouteRow, agent: string, claim: "other" | "unclaimed"): boolean { + return claim === "other" ? Boolean(row.claimed_by && row.claimed_by !== agent) : !row.claimed_by; +} +function compact(row: RouteRow, gate = false): JsonObject { + const result: JsonObject = {}; + for (const field of ["todo_id", "claimed_by", "unblocks_todo_id"] as const) { + if (row[field]) result[field] = row[field]; + } + if (row.excluded_agents.length) result.excluded_agents = [...row.excluded_agents]; + if (gate) { + if (row.gate_state != null) result.gate_state = row.gate_state; + } else { + if (row.blocks_agent) result.blocks_agent = row.blocks_agent; + for (const field of ["priority", "action_kind"] as const) { + if (row[field] != null) result[field] = row[field]; + } + if (!row.todo_id) result.text = row.display_text; + } + return result; +} +function candidates(source: (RouteRow | null)[] | null, agent: string, claim: "other" | "unclaimed"): JsonObject[] { + const result: JsonObject[] = [], seen = new Set(); + for (const row of source ?? []) { + if (!row || !actionable(row) || !matchesClaim(row, agent, claim) || !row.text) continue; + const identity = row.todo_id || row.display_text; + if (seen.has(identity)) continue; + seen.add(identity); + result.push(compact(row)); + if (result.length >= ITEM_LIMIT) break; + } + return result; +} +function countRows(source: (RouteRow | null)[] | null, agent: string, claim: "other" | "unclaimed"): number { + return (source ?? []).filter((row) => row && actionable(row) && matchesClaim(row, agent, claim)).length; +} +export function projectGoalRouteHint(value: unknown): JsonObject | null { + const facts = decodeGoalRouteContext(value), agent = facts.agent_id; + if (!agent) return null; + const summary = facts.agent_todo_summary, lane = facts.agent_lane_next_action; + let current: JsonObject | null = null; + if (lane && (lane.todo_id || lane.text)) { + current = lane.todo_id ? { todo_id: lane.todo_id } : { text: lane.text }; + for (const field of ["selected_by", "confidence", "source"] as const) { + if (lane[field] != null) current[field] = lane[field]; + } + if (lane.claim_required_before_work) current.claim_required_before_work = true; + } + let other = candidates(summary.claimed_advancement_open_items, agent, "other"); + if (!other.length) other = candidates(summary.claimed_by_others_items, agent, "other"); + const source = summary.executable_backlog_items ?? summary.unclaimed_priority_open_items; + const unclaimed = candidates(source, agent, "unclaimed"); + const gates: JsonObject[] = [], seenGates = new Set(); + for (const row of summary.current_agent_handoff_gates ?? summary.handoff_gates ?? []) { + if (!row || !row.excluded_agents.includes(agent) || row.gate_state !== "blocking" || seenGates.has(row.gate_identity)) continue; + seenGates.add(row.gate_identity); + gates.push(compact(row, true)); + if (gates.length >= ITEM_LIMIT) break; + } + const currentCount = summary.current_agent_claimed_advancement_count; + const otherCount = Math.max(other.length, summary.claimed_advancement_open_count + ? Math.max(0, summary.claimed_advancement_open_count - currentCount) + : countRows(summary.claimed_advancement_open_items, agent, "other")); + let decision = "durable_goal_route", reason = "no agent-lane override is needed; preserve durable goal Next Action"; + if (current) { + if (current.claim_required_before_work === true) { + decision = "claim_then_run_current_agent"; + reason = "quota selected an unclaimed in-scope todo for this agent lane"; + } else { + decision = "run_current_agent_lane"; + reason = "quota selected a runnable current-agent lane todo"; + } + } else if (facts.agent_scope_frontier !== null) { + decision = facts.agent_scope_frontier.action || facts.agent_scope_frontier.effective_action || "agent_scope_frontier"; + reason = facts.agent_scope_frontier.reason || "agent-scope frontier blocks delivery"; + } else if (gates.length || other.length) { + decision = "wait_or_reassign_other_agent_lane"; + reason = "visible advancement work is owned by another agent lane"; + } else if (unclaimed.length) { + decision = "claim_unowned_in_scope"; + reason = "unclaimed advancement work is available to the current agent lane"; + } + const hint: JsonObject = { + schema_version: GOAL_ROUTE_HINT_SCHEMA_VERSION, + kind: "agent_lane_synthesis", agent_id: agent, agent_model: "peer_v1", + route_decision: decision, reason, + preserves_goal_next_action: true, goal_next_action_mutation: "none", + counts: { + current_agent_claimed_advancement_count: currentCount, + unclaimed_advancement_count: countRows(source, agent, "unclaimed"), + other_agent_claimed_advancement_count: otherCount, + blocking_handoff_gate_count: gates.length, + }, + selected_action_differs_from_durable: Boolean(facts.active_state_next_action && facts.selected_recommended_action && !facts.actions_projection_aligned), + has_durable_next_action: Boolean(facts.active_state_next_action), + has_latest_run_recommended_action: Boolean(facts.latest_run_recommended_action), + }; + if (current) hint.current_agent_next_action = current; + if (other.length) hint.other_agent_next_actions = other; + if (unclaimed.length) hint.unclaimed_next_actions = unclaimed; + if (gates.length) hint.blocking_handoff_gates = gates; + if (facts.agent_lane_frontier_hint !== null) { + const frontier: JsonObject = {}; + for (const field of ["decision", "source", "reason_code", "target_todo_id", "quiet_noop_allowed"] as const) { + if (facts.agent_lane_frontier_hint[field] != null) frontier[field] = facts.agent_lane_frontier_hint[field]; + } + hint.frontier_hint = frontier; + } + const patch = facts.goal_route_hint_override; + if (patch) { + if (patch.omit_current_agent_next_action === true) delete hint.current_agent_next_action; + for (const [field, value] of Object.entries(patch)) { + if (field !== "omit_current_agent_next_action") hint[field] = value; + } + } + return hint; +} diff --git a/scripts/generate_coordination_state_contract.py b/scripts/generate_coordination_state_contract.py index ec494cbde8..12c488528c 100644 --- a/scripts/generate_coordination_state_contract.py +++ b/scripts/generate_coordination_state_contract.py @@ -135,6 +135,8 @@ "selection_result_schema", "planning_packet_request_schema", "planning_packet_result_schema", + "planning_packet_request_v1_schema", + "planning_packet_result_v1_schema", ) TODO_RESUME_PROTOCOL_KEYS = ( "normalize_request_schema", diff --git a/tests/control_plane/test_agent_scope_frontier_contract.py b/tests/control_plane/test_agent_scope_frontier_contract.py index 35bdba024d..972a44057e 100644 --- a/tests/control_plane/test_agent_scope_frontier_contract.py +++ b/tests/control_plane/test_agent_scope_frontier_contract.py @@ -8,6 +8,7 @@ import sys import pytest +from tests.control_plane.cli_output_fixture import read_cli_context_for_oracle from canonical_authority_fixture import advance_canonical_fixture_authority, read_fixture_todos from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection @@ -23,7 +24,6 @@ quota_status_payload, quota_todo_item, ) -from loopx.control_plane.work_items.goal_route_hint import build_goal_route_hint from loopx.quota import build_quota_should_run, render_quota_should_run_markdown @@ -69,59 +69,6 @@ def test_frontier_keeps_bounded_candidate_context(): assert payload["priority_preemption"] is True -@pytest.mark.parametrize( - "frontier, expected", - [ - ( - { - "schema_version": "agent_scope_frontier_v0", - "effective_action": "agent_scope_wait", - }, - "agent_scope_wait", - ), - ( - { - "schema_version": "agent_scope_frontier_v0", - "action": "agent_scope_wait", - "effective_action": "agent_scope_wait", - }, - "agent_scope_wait", - ), - ( - { - "schema_version": "agent_scope_frontier_v1", - "action": "successor_replan_required", - }, - "successor_replan_required", - ), - ( - { - "schema_version": "agent_scope_frontier_v1", - "action": "successor_replan_required", - "effective_action": "agent_scope_wait", - }, - "successor_replan_required", - ), - ], -) -def test_goal_route_reads_canonical_action_first_and_legacy_as_fallback( - frontier, expected -): - hint = build_goal_route_hint( - agent_identity={"agent_id": "frontier-fixture"}, - agent_todo_summary={}, - agent_lane_next_action=None, - agent_scope_frontier=frontier, - agent_lane_frontier_hint=None, - active_state_next_action="Preserve the goal route.", - latest_run_recommended_action=None, - selected_recommended_action="Wait.", - ) - assert hint["route_decision"] == expected - assert hint["preserves_goal_next_action"] is True - assert hint["goal_next_action_mutation"] == "none" - - def test_live_should_run_and_markdown_use_the_v1_frontier(): status = quota_status_payload( goal_id="frontier-fixture", @@ -196,7 +143,8 @@ def test_cli_should_run_reads_disposable_state_and_emits_v1(tmp_path): text=True, check=True, ) - guard = json.loads(result.stdout) + captured, _reader_cost = read_cli_context_for_oracle(result.stdout) + guard = json.loads(captured) assert guard["effective_action"] == "reassignment_required" assert guard["should_run"] is False assert guard["agent_scope_frontier"]["schema_version"] == "agent_scope_frontier_v1" diff --git a/tests/control_plane/test_closed_monitor_replay.py b/tests/control_plane/test_closed_monitor_replay.py index 85cd58c06b..9d010d1b5b 100644 --- a/tests/control_plane/test_closed_monitor_replay.py +++ b/tests/control_plane/test_closed_monitor_replay.py @@ -4,6 +4,7 @@ import json import pytest +from tests.control_plane.cli_output_fixture import read_cli_context_for_oracle from canonical_authority_fixture import isolate_sqlite_runtime from loopx.control_plane.testing.canary_harness import run_json_cli, run_json_cli_result @@ -25,7 +26,9 @@ def test_closed_monitor_replays_original_no_spend_turn( target_key="public-target", next_due_at="2000-01-01T00:00:00Z") def call(*args): - return run_json_cli(*args, registry_path=registry, runtime_root=runtime) + wire = run_json_cli(*args, registry_path=registry, runtime_root=runtime) + captured, _reader_cost = read_cli_context_for_oracle(json.dumps(wire)) + return json.loads(captured) turn = "closed-monitor-observation" guard = ["quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, @@ -53,6 +56,7 @@ def call(*args): for selector in ([], ["--todo-id", monitor["todo_id"]]): replay = call(*guard, *selector) assert replay["heartbeat_receipt"]["settlement_identity"] == identity + assert "goal_route_hint" not in replay assert replay["should_run"] is False assert replay["effective_action"] == "heartbeat_settled_skip" assert not replay["interaction_contract"]["agent_channel"]["must_attempt"] diff --git a/tests/control_plane/test_effect_turn_live_quota_decision.py b/tests/control_plane/test_effect_turn_live_quota_decision.py index c70e1c7ca1..292ba9479d 100644 --- a/tests/control_plane/test_effect_turn_live_quota_decision.py +++ b/tests/control_plane/test_effect_turn_live_quota_decision.py @@ -297,6 +297,7 @@ def test_owner_cadence_wait_is_app_only_and_has_no_scheduler_side_effect( "eligible_now": False, "next_eligible_at_ms": next_eligible_at_ms, } + assert "goal_route_hint" not in packet assert packet["decision"] == "wait" assert packet["effective_action"] == "blocked_wait" assert packet["should_run"] is False diff --git a/tests/control_plane/test_goal_route_planning_adapter.py b/tests/control_plane/test_goal_route_planning_adapter.py new file mode 100644 index 0000000000..554269b57c --- /dev/null +++ b/tests/control_plane/test_goal_route_planning_adapter.py @@ -0,0 +1,129 @@ +"""The adapter snapshots compatibility facts; one native batch owns routing.""" +from copy import deepcopy + +import pytest + +from loopx.control_plane.work_items import action_portfolio as adapter +from loopx.control_plane.testing.quota_fixtures import quota_status_payload, quota_todo_item +from loopx.quota import build_quota_should_run + + +def _context(**kwargs): + return adapter.capture_goal_route_context(**{ + "agent_identity": {"agent_id": "route-agent"}, "agent_todo_summary": {}, + "agent_lane_next_action": None, "agent_scope_frontier": None, + "agent_lane_frontier_hint": None, "active_state_next_action": "Durable step", + "latest_run_recommended_action": None, "selected_recommended_action": "Durable step", + "goal_route_hint_override": None, **kwargs, + }) + + +def _packet(**kwargs): + return adapter.build_quota_planning_packet(**{ + "projection_enabled": False, "include_detail": False, "goal_id": "route-fixture", + "selected": None, "agent_id": "route-agent", "agent_todo_summary": {}, + "agent_todo_source_items": [], "capability_gate": None, + "blocked_priority_fallback": None, "goal_frontier_projection": None, **kwargs, + }) + + +def test_route_only_uses_one_batch_without_inventory_or_goal_context(monkeypatch): + from loopx.control_plane.work_items import replan_context_codec + + def forbidden(*args, **kwargs): + pytest.fail("route-only must not read inventory or Goal context") + + monkeypatch.setattr(adapter, "build_quota_planning_inventory_request", forbidden) + monkeypatch.setattr(replan_context_codec, "goal_context_facts", forbidden) + native = adapter.effect_runtime_result + calls = [] + + def observed(operation, request): + calls.append((operation, deepcopy(request))) + return native(operation, request) + + monkeypatch.setattr(adapter, "effect_runtime_result", observed) + assert _packet() == {} + assert calls == [] + result = _packet(goal_route_context=_context(), source_status={"registry": {"irrelevant": True}}) + assert result["goal_route_hint"]["route_decision"] == "durable_goal_route" + assert len(calls) == 1 + assert calls[0][0] == "work_item.action_portfolio.project" + assert calls[0][1]["planning"] is None + + +def test_capture_preserves_empty_sources_all_rows_and_snapshot_identity(): + text = "\u2003long🙂 " * 100 + summary = { + "executable_backlog_items": [], + "unclaimed_priority_open_items": [{"text": text}] * 5, + "current_agent_handoff_gates": [], "handoff_gates": "invalid", + } + frontier = {} + override = {"kind": "task_orchestration", "peer_lane_count": 2} + context = _context(agent_todo_summary=summary, agent_scope_frontier=frontier, + goal_route_hint_override=override) + facts = context["agent_todo_summary"] + assert facts["executable_backlog_items"] == [] + assert facts["current_agent_handoff_gates"] == [] + assert facts["handoff_gates"] is None + assert facts["claimed_by_others_items"] is None + assert len(facts["unclaimed_priority_open_items"]) == 5 + assert len(facts["unclaimed_priority_open_items"][0]["display_text"]) <= 120 + assert context["agent_scope_frontier"] == {"action": "", "effective_action": "", "reason": ""} + assert context["actions_projection_aligned"] is True + summary["executable_backlog_items"].append({"text": "later mutation"}) + override["peer_lane_count"] = 99 + assert facts["executable_backlog_items"] == [] + assert context["goal_route_hint_override"]["peer_lane_count"] == 2 + + +def test_invalid_agent_does_not_create_override_only_route(): + assert _context(agent_identity={}, goal_route_hint_override={"kind": "task_orchestration"}) is None + + +def test_monitor_only_removes_native_batch_route_before_public_readback(): + status = quota_status_payload( + goal_id="route-fixture", status="active", recommended_action="Complete the deliverable", + coordination={"registered_agents": ["route-agent"], + "agent_work_modes": {"route-agent": "monitor_only"}}, + agent_todo_items=[quota_todo_item(todo_id="todo_route_work", title="Complete the deliverable", + claimed_by="route-agent")], + ) + result = build_quota_should_run(status, goal_id="route-fixture", agent_id="route-agent") + assert result["should_run"] is False + assert result["agent_work_mode"] == "monitor_only" + assert "goal_route_hint" not in result + + +def test_planning_and_route_share_one_existing_rpc(monkeypatch): + native = adapter.effect_runtime_result + calls = [] + + def observed(operation, request): + calls.append((operation, deepcopy(request))) + return native(operation, request) + + monkeypatch.setattr(adapter, "effect_runtime_result", observed) + candidate = {"todo_id": "todo_route_work", "text": "Complete the deliverable", + "status": "open", "task_class": "advancement_task"} + result = _packet(projection_enabled=True, selected=candidate, + goal_route_context=_context(agent_lane_next_action=candidate)) + assert result["action_portfolio"] + assert result["goal_route_hint"]["route_decision"] == "run_current_agent_lane" + assert len(calls) == 1 + assert calls[0][1]["planning"]["planning_inventory_request"] + + +def test_compatibility_text_and_gate_identity_use_python_owner(): + text = "\u001c\u2003" + "🙂" * 240 + "\u00a0tail" + context = _context(agent_todo_summary={"handoff_gates": [{ + "index": 0, "text": text, "todo_id": None, "excluded_agents": ("route-agent",), + "gate_state": "blocking", + }]}, selected_recommended_action=False) + row = context["agent_todo_summary"]["handoff_gates"][0] + assert row["gate_identity"] == text + assert row["text"] == "🙂" * 219 + "…" + assert row["display_text"] == "🙂" * 119 + "…" + assert row["excluded_agents"] == ["route-agent"] + assert context["selected_recommended_action"] == "" diff --git a/tests/control_plane/test_quota_run_decision.py b/tests/control_plane/test_quota_run_decision.py index 33481d7043..e56b9013e9 100644 --- a/tests/control_plane/test_quota_run_decision.py +++ b/tests/control_plane/test_quota_run_decision.py @@ -15,7 +15,6 @@ QuotaRunDecision, resolve_quota_run_decision, ) -from loopx.control_plane.quota.task_orchestration import apply_task_route_hint_override def _resolve(**overrides: Any) -> QuotaRunDecision: @@ -189,26 +188,6 @@ def test_task_orchestration_refines_normal_run_action() -> None: assert decision.goal_route_hint_override["peer_lane_count"] == 2 -def test_goal_route_hint_applies_only_the_typed_owner_patch() -> None: - hint = {"kind": "agent_lane_synthesis", "current_agent_next_action": {"todo_id": "todo-a"}, - "has_durable_next_action": True} - patched = apply_task_route_hint_override(hint, { - "omit_current_agent_next_action": True, - "kind": "task_orchestration", - "route_decision": "coordinate_task_bundle", - "reason": "task-scoped coordinator may spawn admitted child lanes", - "peer_lane_count": 2, - }) - assert patched == { - "kind": "task_orchestration", - "route_decision": "coordinate_task_bundle", - "reason": "task-scoped coordinator may spawn admitted child lanes", - "peer_lane_count": 2, - "has_durable_next_action": True, - } - assert hint["current_agent_next_action"] == {"todo_id": "todo-a"} - - @pytest.mark.parametrize( "reply,material,action,reason", [ diff --git a/tests/control_plane/test_settled_replay_construction.py b/tests/control_plane/test_settled_replay_construction.py index 63ebcfd6fc..84632d85cf 100644 --- a/tests/control_plane/test_settled_replay_construction.py +++ b/tests/control_plane/test_settled_replay_construction.py @@ -104,6 +104,8 @@ def unexpected(*args, **kwargs): pytest.fail("settled replay entered an executable action construction path") monkeypatch.setattr(should_run_packet, "_resolve_agent_lane_delivery_route", unexpected) + monkeypatch.setattr(should_run_packet, "capture_goal_route_context", unexpected) + monkeypatch.setattr(should_run_packet, "build_quota_planning_packet", unexpected) monkeypatch.setattr(should_run_packet, "build_replan_action_packet", unexpected) monkeypatch.setattr(should_run_packet, "_apply_agent_monitor_only_precedence", unexpected) status = quota_status_payload( @@ -135,6 +137,7 @@ def unexpected(*args, **kwargs): "autonomous_replan_obligation", "action_portfolio", "protocol_action_packet", + "goal_route_hint", ): assert field not in payload assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is False diff --git a/tests/control_plane/test_unadmitted_selection_construction.py b/tests/control_plane/test_unadmitted_selection_construction.py index a1db89e355..46ee7819da 100644 --- a/tests/control_plane/test_unadmitted_selection_construction.py +++ b/tests/control_plane/test_unadmitted_selection_construction.py @@ -322,6 +322,7 @@ def test_scoped_fallback_cannot_execute_under_an_unadmitted_selection(): assert payload["effective_action"] == "quota_skip" assert payload["state"] == "quota_action_selection_rejected" + assert "goal_route_hint" not in payload for flag in ( "ok", "should_run", "actionable_by_codex", "safe_bypass_allowed", "spend_allowed_now", "spend_after_validation", "normal_delivery_allowed", diff --git a/tests/control_plane_ts/action_portfolio.test.ts b/tests/control_plane_ts/action_portfolio.test.ts index 796b19b065..f6580e025f 100644 --- a/tests/control_plane_ts/action_portfolio.test.ts +++ b/tests/control_plane_ts/action_portfolio.test.ts @@ -589,3 +589,55 @@ test("retained explicit selection cannot be replaced by a projected default", () }, }); }); + +function routeFacts() { + return { + agent_id: "agent", agent_lane_next_action: null, agent_scope_frontier: {}, agent_lane_frontier_hint: null, + agent_todo_summary: { + claimed_advancement_open_items: null, claimed_by_others_items: null, + executable_backlog_items: [], unclaimed_priority_open_items: null, + current_agent_handoff_gates: [], handoff_gates: null, + current_agent_claimed_advancement_count: 0, claimed_advancement_open_count: 0, + }, + active_state_next_action: "durable", selected_recommended_action: "", latest_run_recommended_action: "", + actions_projection_aligned: true, goal_route_hint_override: null, + }; +} + +test("v1 route-only batch needs no selected Todo, inventory or Goal context", () => { + const request = {schema_version: "quota_planning_packet_request_v1", planning: null, goal_route_context: routeFacts()}; + const original = structuredClone(request); + const result = projectQuotaActionPortfolio(request)!; + assert.equal(result.schema_version, "quota_planning_packet_v1"); + assert.equal((result.goal_route_hint as Record).route_decision, "agent_scope_frontier"); + assert.equal(result.action_portfolio, undefined); + assert.equal(result.agent_todo_planning_inventory, undefined); + assert.deepEqual(request, original); + assert.deepEqual(projectQuotaActionPortfolio({...request, goal_route_context: null}), {schema_version: "quota_planning_packet_v1"}); +}); + +test("v1 combines planning and routing using exactly the v0 planning projection", () => { + const primary = candidate("todo_primary001", "Run primary", "P0"); + const legacy = planningPacketRequest(primary, [candidate("todo_other0001", "Run other", "P1")]); + const legacyResult = projectQuotaActionPortfolio(legacy)!; + const {schema_version: ignored, ...planning} = legacy; + const context = routeFacts(); + const combined = projectQuotaActionPortfolio({schema_version: "quota_planning_packet_request_v1", planning, goal_route_context: context})!; + const {schema_version, goal_route_hint, ...views} = combined; + const {schema_version: oldVersion, ...oldViews} = legacyResult; + assert.equal(schema_version, "quota_planning_packet_v1"); + assert.deepEqual(views, oldViews); + assert.ok(goal_route_hint); + assert.deepEqual(projectQuotaActionPortfolio({schema_version: "quota_planning_packet_request_v1", planning, goal_route_context: null}), + {...legacyResult, schema_version: "quota_planning_packet_v1"}); +}); + +test("v1 rejects omitted blocks, bad block types, malformed route facts and unsupported versions", () => { + const base = {schema_version: "quota_planning_packet_request_v1", planning: null, goal_route_context: null}; + for (const request of [ + {schema_version: base.schema_version}, {...base, planning: []}, {...base, planning: {}}, + {...base, goal_route_context: []}, {...base, goal_route_context: {}}, + {...base, schema_version: "quota_planning_packet_request_v99"}, + {schema_version: QUOTA_PLANNING_PACKET_REQUEST_SCHEMA_VERSION, goal_route_context: routeFacts()}, + ]) assert.throws(() => projectQuotaActionPortfolio(request)); +}); diff --git a/tests/control_plane_ts/goal_route_hint.test.ts b/tests/control_plane_ts/goal_route_hint.test.ts new file mode 100644 index 0000000000..7eed817602 --- /dev/null +++ b/tests/control_plane_ts/goal_route_hint.test.ts @@ -0,0 +1,176 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { projectGoalRouteHint } from "../../loopx/control_plane/work_items/goal_route_hint.ts"; + +// These are transport facts, not pre-selected routes. Python-only text/normalizer +// compatibility is covered at the adapter boundary, including Unicode truncation. +export function routeContext(overrides: Record = {}) { + return { + agent_id: "current-agent", + agent_todo_summary: { + claimed_advancement_open_items: null, + claimed_by_others_items: null, + executable_backlog_items: null, + unclaimed_priority_open_items: null, + current_agent_handoff_gates: null, + handoff_gates: null, + current_agent_claimed_advancement_count: 0, + claimed_advancement_open_count: 0, + }, + agent_lane_next_action: null, + agent_scope_frontier: null, + agent_lane_frontier_hint: null, + active_state_next_action: "Durable next action", + selected_recommended_action: "Selected action", + latest_run_recommended_action: "", + actions_projection_aligned: false, + goal_route_hint_override: null, + ...overrides, + }; +} +function row(overrides: Record = {}) { + return {done: false, status: "open", task_class: "advancement_task", + todo_id: null, claimed_by: null, blocks_agent: null, unblocks_todo_id: null, + excluded_agents: [], text: "Do work", display_text: "Do work", gate_identity: "gate", gate_state: null, + ...overrides}; +} +function contextWithSummary(summary: Record) { + const context = routeContext(); + return {...context, agent_todo_summary: {...context.agent_todo_summary, ...summary}}; +} +function current(claim = false) { + return {todo_id: "todo_current", text: "", claim_required_before_work: claim}; +} + +test("current action outranks frontier, peers and unclaimed work, with ID-only current actions", () => { + for (const claim of [false, true]) { + const context = contextWithSummary({claimed_advancement_open_items: [row({claimed_by: "peer"})], + executable_backlog_items: [row()]}); + const hint = projectGoalRouteHint({...context, agent_lane_next_action: current(claim), + agent_scope_frontier: {action: "wait", reason: "blocked"}})!; + assert.equal(hint.route_decision, claim ? "claim_then_run_current_agent" : "run_current_agent_lane"); + assert.deepEqual(hint.current_agent_next_action, {todo_id: "todo_current", ...(claim ? {claim_required_before_work: true} : {})}); + } +}); + +test("canonical frontier action wins; legacy action and empty frontier retain their distinct meaning", () => { + for (const [frontier, decision] of [ + [{effective_action: "agent_scope_wait"}, "agent_scope_wait"], + [{action: "successor_replan_required"}, "successor_replan_required"], + [{action: "successor_replan_required", effective_action: "agent_scope_wait"}, "successor_replan_required"], + [{}, "agent_scope_frontier"], [null, "durable_goal_route"], + ] as const) { + const hint = projectGoalRouteHint(routeContext({agent_scope_frontier: frontier}))!; + assert.equal(hint.route_decision, decision); + assert.equal(hint.preserves_goal_next_action, true); + assert.equal(hint.goal_next_action_mutation, "none"); + } +}); + +test("present empty primary lists suppress stale fallback candidates and gates (existence mutation guard)", () => { + const backup = row({excluded_agents: ["current-agent"], gate_state: "blocking"}); + const context = contextWithSummary({executable_backlog_items: [], unclaimed_priority_open_items: [backup], + current_agent_handoff_gates: [], handoff_gates: [backup]}); + const hint = projectGoalRouteHint(context)!; + assert.equal(hint.route_decision, "durable_goal_route"); + assert.equal(hint.unclaimed_next_actions, undefined); + assert.equal(hint.blocking_handoff_gates, undefined); + assert.deepEqual(hint.counts, {current_agent_claimed_advancement_count: 0, + unclaimed_advancement_count: 0, other_agent_claimed_advancement_count: 0, blocking_handoff_gate_count: 0}); + const fallback = projectGoalRouteHint(contextWithSummary({unclaimed_priority_open_items: [backup], handoff_gates: [backup]}))!; + assert.equal(fallback.route_decision, "wait_or_reassign_other_agent_lane"); + assert.equal(fallback.current_agent_next_action, undefined); +}); + +test("peer selection fallback uses selected results, while counts use the original full source", () => { + const context = contextWithSummary({claimed_advancement_open_items: [row({claimed_by: "peer", text: ""})], + claimed_by_others_items: [row({claimed_by: "peer", todo_id: "other"})]}); + const hint = projectGoalRouteHint(context)!; + assert.equal(hint.route_decision, "wait_or_reassign_other_agent_lane"); + assert.deepEqual(hint.other_agent_next_actions, [{todo_id: "other", claimed_by: "peer"}]); + assert.equal(hint.current_agent_next_action, undefined); +}); + +test("ordered display deduplicates to three but counts duplicates, empty text, and all actionable rows", () => { + const source = [null, row({done: true}), row({status: "blocked"}), row({task_class: "continuous_monitor"}), + row({claimed_by: "peer"}), row({todo_id: "a", text: ""}), row({todo_id: "a"}), row({todo_id: "a"}), + row({text: "Same visible prefix with first long suffix", display_text: "No ID", blocks_agent: "blocked", priority: 0, action_kind: "implement"}), + row({text: "Same visible prefix with different long suffix", display_text: "No ID"}), row({todo_id: "b"}), row({todo_id: "c"})]; + const hint = projectGoalRouteHint(contextWithSummary({executable_backlog_items: source}))!; + assert.equal(hint.route_decision, "claim_unowned_in_scope"); + assert.deepEqual(hint.unclaimed_next_actions, [{todo_id: "a"}, + {text: "No ID", blocks_agent: "blocked", priority: 0, action_kind: "implement"}, {todo_id: "b"}]); + assert.deepEqual(hint.counts, {current_agent_claimed_advancement_count: 0, + unclaimed_advancement_count: 7, other_agent_claimed_advancement_count: 0, blocking_handoff_gate_count: 0}); +}); + +test("other counts preserve supplied-total subtraction, zero fallback and displayed lower bound", () => { + for (const [total, mine, expected] of [[10, 3, 7], [0, 8, 5], [1, 9, 3]]) { + const hint = projectGoalRouteHint(contextWithSummary({claimed_advancement_open_count: total, + current_agent_claimed_advancement_count: mine, + claimed_advancement_open_items: Array.from({length: 5}, (_, i) => row({claimed_by: "peer", todo_id: `other${i}`}))}))!; + assert.equal((hint.counts as Record).other_agent_claimed_advancement_count, expected); + } +}); + +test("blocking gates use exclusion and original identity, not candidate eligibility, capped at three", () => { + const gate = (id: string, extra = {}) => row({todo_id: id, gate_identity: id, done: true, + text: "", status: "closed", excluded_agents: ["current-agent"], gate_state: "blocking", ...extra}); + const hint = projectGoalRouteHint(contextWithSummary({handoff_gates: [null, + gate("ignore", {excluded_agents: ["someone"]}), gate("ignore2", {gate_state: "cleared"}), + gate("a"), gate("a"), gate("b"), gate("c"), gate("d")]}))!; + assert.deepEqual(hint.blocking_handoff_gates, ["a", "b", "c"].map(todo_id => ({todo_id, excluded_agents: ["current-agent"], gate_state: "blocking"}))); + assert.equal((hint.counts as Record).blocking_handoff_gate_count, 3); +}); + +test("typed task override removes current action before merging and never creates a missing-agent hint", () => { + const patch = {omit_current_agent_next_action: true, kind: "task_orchestration", route_decision: "coordinate_task_bundle", + reason: "task-scoped coordinator may spawn admitted child lanes", peer_lane_count: 2}; + const context = routeContext({agent_lane_next_action: current(), goal_route_hint_override: patch}); + const before = structuredClone(context); + const hint = projectGoalRouteHint(context)!; + for (const field of ["kind", "route_decision", "reason", "peer_lane_count"] as const) assert.equal(hint[field], patch[field]); + assert.equal(hint.current_agent_next_action, undefined); + assert.equal(hint.omit_current_agent_next_action, undefined); + assert.equal(hint.has_durable_next_action, true); + assert.deepEqual(context, before); + assert.equal(projectGoalRouteHint({...context, agent_id: null}), null); + assert.equal(projectGoalRouteHint({...context, agent_id: ""}), null); + assert.deepEqual(projectGoalRouteHint({...context, goal_route_hint_override: {omit_current_agent_next_action: false}})!.current_agent_next_action, {todo_id: "todo_current"}); +}); + +test("frontier metadata omission and named alignment compatibility remain read-only", () => { + const context = routeContext({agent_lane_next_action: {...current(), todo_id: null, text: "text only", source: null, selected_by: "lane", confidence: 0}, + agent_lane_frontier_hint: {decision: "wait", source: null, reason_code: "cause", quiet_noop_allowed: false, ignored: "x"}, + latest_run_recommended_action: "latest", actions_projection_aligned: true}); + const before = structuredClone(context), hint = projectGoalRouteHint(context)!; + assert.equal(hint.selected_action_differs_from_durable, false); + assert.equal(hint.has_latest_run_recommended_action, true); + assert.deepEqual(hint.frontier_hint, {decision: "wait", reason_code: "cause", quiet_noop_allowed: false}); + assert.deepEqual(hint.current_agent_next_action, {text: "text only", selected_by: "lane", confidence: 0}); + assert.deepEqual(context, before); + assert.deepEqual(projectGoalRouteHint(routeContext({agent_lane_frontier_hint: {}}))!.frontier_hint, {}); +}); + +test("malformed routing facts are rejected instead of cast into domain types", () => { + for (const context of [null, {}, routeContext({agent_id: 3}), routeContext({actions_projection_aligned: "false"}), + routeContext({agent_lane_next_action: {}}), routeContext({agent_scope_frontier: []}), + routeContext({goal_route_hint_override: {peer_lane_count: -1}}), routeContext({goal_route_hint_override: {arbitrary: true}}), + contextWithSummary({executable_backlog_items: {}}), contextWithSummary({executable_backlog_items: [{}]}), + contextWithSummary({current_agent_claimed_advancement_count: -1})]) { + assert.throws(() => projectGoalRouteHint(context)); + } +}); + + +test("gate deduplication retains distinct raw identities even when normalized Todo IDs coincide", () => { + const base = row({todo_id: "same-id", excluded_agents: ["current-agent"], gate_state: "blocking"}); + const hint = projectGoalRouteHint(contextWithSummary({handoff_gates: [ + {...base, gate_identity: " same-id "}, {...base, gate_identity: "same-id"}, + {...base, gate_identity: "same-id"}, + ]}))!; + assert.deepEqual(hint.blocking_handoff_gates, [ + {todo_id: "same-id", excluded_agents: ["current-agent"], gate_state: "blocking"}, + {todo_id: "same-id", excluded_agents: ["current-agent"], gate_state: "blocking"}, + ]); +}); diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index b211b87e2b..91a31c52c7 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -37,6 +37,8 @@ "tests/control_plane_ts/effect_runtime_errors.test.ts", "tests/control_plane_ts/post_writeback_hook_transaction.test.ts", "tests/control_plane_ts/action_review_plan.test.ts", + "tests/control_plane_ts/goal_route_hint.test.ts", + "tests/control_plane_ts/action_portfolio.test.ts", "tests/control_plane_ts/goal_workspace_digest.test.ts", "tests/control_plane_ts/interaction_contract.test.ts", "tests/control_plane_ts/scheduler_arbitration.test.ts",