From a81fc32ee62537a5861693248671f069366f810e Mon Sep 17 00:00:00 2001 From: aihop <631479666@qq.com> Date: Fri, 6 Nov 2026 08:14:38 +0800 Subject: [PATCH 1/2] refactor(agents): project peer directory in TypeScript --- .../2026-11-06-peer-directory.md | 71 ++++ .../2026-11-06-peer-directory.zh-CN.md | 56 +++ .../typescript-control-plane-migration-v0.md | 1 + ...script-control-plane-migration-v0.zh-CN.md | 1 + .../contracts/projection-envelope-contract.md | 1 + ...peer-agent-directory-and-observation-v0.md | 45 ++- loopx/cli_commands/agent_directory.py | 6 +- loopx/control_plane/agents/directory.py | 325 +++++------------- loopx/control_plane/agents/directory.ts | 142 ++++++++ .../control_plane/effect_runtime_handlers.ts | 1 + .../test_peer_agent_directory.py | 97 +++++- .../peer_agent_directory.test.ts | 101 ++++++ tsconfig.control-plane.json | 1 + 13 files changed, 591 insertions(+), 257 deletions(-) create mode 100644 docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md create mode 100644 docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md create mode 100644 loopx/control_plane/agents/directory.ts create mode 100644 tests/control_plane_ts/peer_agent_directory.test.ts diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md new file mode 100644 index 0000000000..5e8e7f397d --- /dev/null +++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md @@ -0,0 +1,71 @@ +# Peer directory projection and inherited source evidence + +## Outcome and ownership + +This bounded T3 projection migration removes Python's second directory decision +owner. The public `agent-directory` caller keeps its command, handler and +`build_peer_agent_directory()` signature. One `agent.directory.project` request +owns membership, work fields, route publication caps, counts, cursor, +limitations and page-local rollup, and invokes `sealProjectionEnvelope` directly +in the same TypeScript process. + +Python retains source collection, public-output safety checks and request/result +compatibility. Its `_work_block()`, `_rollup()` and replaced directory assembly +branches are retired, with no semantic fallback on bridge failure. The existing +management projection remains the owner of Todo selection and management state; +the existing binding owner remains the route interpreter. Neither is copied +into the directory projector. + +## Compatibility and authority + +Membership uses the complete registered set, including Agents outside the +24-row display page. Unregistered historical claimants remain diagnostic rows, +not members. `omitted_row_count` keeps its whole-directory definition on later +and empty pages. Rollup category counts cover the returned page before its +eight-item display cap. Route counts include withheld candidates even though +at most three publishable candidates are shown. Unavailable, empty and partial +execution facts retain their distinct meanings, as do false, null and absent +fields. + +The only intended public additions are `projection_envelope` and necessary +source-warning presentation. The envelope inherits upstream source observations +rather than replacing them with directory assembly time. Missing evidence stays +unknown; source incompleteness is distinct from display pagination. This does +not create a presence provider, resolve an exact host route, confer execution +permission or alter claims, leases, persistence or provider defaults. See the +[directory contract](../../../../reference/protocols/peer-agent-directory-and-observation-v0.md#directory-projection-ownership-and-source-evidence). + +## Cost and acceptance boundary + +The fixed implementation baseline is `faab5f5c291c5359216a0f944fe63fcfbaacdedc`. +The previous directory assembly was local Python work; this change adds one +coarse-grained runtime request to that stage. It does **not** make the entire +CLI a one-request operation. Rule ownership and retained source evidence are +the intended benefit; moving code alone is not evidence of a speedup or better +whole-Goal outcomes. + +Qualification is pending until recorded against the final integrated candidate: + +- Fixed-time baseline/head comparison through the actual command, preserving + old fields, ordering, diagnostics and exit status, with envelope additions + checked separately. +- File and SQLite first/last pages, later-page members, unauthorized claimants, + explicit runtime roots and unchanged business state. +- Stale, unreadable, incomplete, missing and replayed source evidence; + independent mutations must detect page-only membership and refreshed old + source timestamps. +- Clean wheel and sdist execution without borrowing the source checkout. +- Matched whole-CLI cold/warm p50/p95, runtime request counts and output sizes; + type checking, native/Python regressions and required premerge checks. + +This checklist is not a passed-test claim. Preserve failures and unmeasured +boundaries when adding results; earlier CLI optimization measurements do not +qualify this candidate. This checkpoint does not close T4 or change release +qualification. + +## Bridge exit and rollback + +Delete the remaining Python adapter when actual directory callers and necessary +fact providers can call the native owner directly. Until then it only supplies +facts, public safety and compatibility transport. No new store, dual-write or +persisted migration is introduced; rollback restores the previous code package. diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md new file mode 100644 index 0000000000..fb5fab4fdd --- /dev/null +++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md @@ -0,0 +1,56 @@ +# Peer Agent 目录投影与上游来源依据 + +## 目标与所有权 + +本次有界 T3 投影迁移删除 Python 中第二份目录决策实现。公开 `agent-directory` +保持原命令、handler 和 `build_peer_agent_directory()` 签名。一次 +`agent.directory.project` 请求拥有成员检查、工作字段、路由展示上限、计数、 +cursor、limitations 和当前页 rollup，并在同一 TypeScript 进程内直接调用 +`sealProjectionEnvelope`。 + +Python 保留来源收集、公共输出安全检查及请求／结果兼容校验；删除 `_work_block()`、 +`_rollup()` 和被替换的目录组装分支，bridge 失败时不回退到 Python 语义实现。 +既有管理投影继续拥有 Todo 选择和管理状态，既有绑定 owner 继续解释路由； +目录投影不复制这两者的算法。 + +## 兼容与权限 + +成员资格依据完整注册集，包括不在当前 24 行展示页的 Agent。历史未注册 claimant +仍是诊断行，不取得成员资格。后续页和空页的 `omitted_row_count` 保留整份目录的 +既有定义。rollup 分类计数覆盖当前返回页，不受八项展示上限影响；路由计数包含 +被隐去的候选，即使只展示最多三个可公开候选。execution facts 未取得、已取得空集 +与部分覆盖保持区别，false、null 和字段缺失也保留各自语义。 + +唯一计划内的公开增量是 `projection_envelope` 及必要的来源告警展示。envelope +继承上游来源观察时间，不以目录组装时间替换；缺失依据保持未知，来源不完整与 +展示分页分别记录。本次不增加 presence provider，不解析精确宿主路由，不授予 +执行权限，不改变 claim、lease、持久化格式或 provider 默认值。见 +[目录契约](../../../../reference/protocols/peer-agent-directory-and-observation-v0.md#directory-projection-ownership-and-source-evidence)。 + +## 成本与验收边界 + +固定实现基线为 `faab5f5c291c5359216a0f944fe63fcfbaacdedc`。此前目录组装为 +Python 本地工作，本次为该阶段增加一次粗粒度运行时请求；这**不代表整个 CLI +只有一次请求**。收益是目录规则收口及保留来源依据，代码迁移本身不能证明加速 +或整 Goal 成果改善。 + +以下资格待最终集成候选的实测结果记录后确认： + +- 实际命令的固定时间 baseline/head 差分，保持旧字段、顺序、诊断及退出状态， + 单独核对 envelope 增量。 +- File／SQLite 首末页、后页成员、未授权 claimant、显式 runtime root，以及 + 业务状态不变。 +- 过期、不可读、不完整、缺失及重放来源依据；独立变异必须能检出只按当前页 + 判断成员以及用当前时间刷新旧来源时间的错误。 +- 不借用源码 checkout 的干净 wheel／sdist 安装执行。 +- 同负载完整 CLI 冷／暖 p50、p95、运行时请求次数及输出体积；类型检查、 + 原生／Python 回归及必需的 premerge 检查。 + +此清单不表示测试已通过。补充结果时须保留失败和未测边界；既往 CLI 优化测量 +不能资格化本候选。本检查点不关闭 T4，也不改变发布资格。 + +## Bridge 退出与回滚 + +当真实目录调用方及必要事实来源可直接调用原生 owner 时，删除剩余 Python +适配器；此前它只负责事实、公共安全与兼容传输。本次不新增存储、双写或持久化 +迁移，回滚仅恢复上一版代码包。 diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index a51674dc62..8761c8003f 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -2856,3 +2856,4 @@ with shared write/read status-phase checks. See the [2026-10-02 checkpoint](ledg - Progress-review receipt qualification: [2026-11-02 checkpoint](ledger/typescript-control-plane-migration-v0/2026-11-02-progress-review-qualification.md). - Global Todo readiness complete owner cutover (both consumers): [2026-11-02 ledger](ledger/typescript-control-plane-migration-v0/2026-11-02-global-todo-readiness.md). - Selected Todo source read inside the context transaction: [2026-11-04](ledger/typescript-control-plane-migration-v0/2026-11-04-selected-context-source-read.md). +- Peer directory projection and inherited source evidence: [2026-11-06 checkpoint](ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md). 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..a13cfcc3e2 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 @@ -2118,3 +2118,4 @@ SQLite 中位数略升，收益不能外推为全面加速。这只资格化本 - 进度评审 receipt 资格判定：[2026-11-02 检查点](ledger/typescript-control-plane-migration-v0/2026-11-02-progress-review-qualification.zh-CN.md)。 - Global Todo 就绪分类完整 owner 切换（两个消费者）：[2026-11-02 记录](ledger/typescript-control-plane-migration-v0/2026-11-02-global-todo-readiness.zh-CN.md)。 - 当前 Todo 来源读取并入上下文事务：[2026-11-04](ledger/typescript-control-plane-migration-v0/2026-11-04-selected-context-source-read.zh-CN.md). +- Peer Agent 目录投影与上游来源依据：[2026-11-06 检查点](ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md)。 diff --git a/docs/reference/contracts/projection-envelope-contract.md b/docs/reference/contracts/projection-envelope-contract.md index 70654b7693..0855dd2ded 100644 --- a/docs/reference/contracts/projection-envelope-contract.md +++ b/docs/reference/contracts/projection-envelope-contract.md @@ -17,6 +17,7 @@ Python-owned projections do so through `loopx/control_plane/projection_envelope_ | Projection | `projection` | Coverage scope | | --- | --- | --- | +| `loopx agent-directory` | `peer_agent_directory` | `goal_registered_agents` | | `loopx status` | `status` | `registry`, `goal` with `--goal-id`, or `activation.` | | `loopx status --use-projection-cache` hit | `status`, `served_from_cache: true` | as stored | | `loopx global-summary` | `global_summary` | `global` | diff --git a/docs/reference/protocols/peer-agent-directory-and-observation-v0.md b/docs/reference/protocols/peer-agent-directory-and-observation-v0.md index 56e0960018..fcdf098000 100644 --- a/docs/reference/protocols/peer-agent-directory-and-observation-v0.md +++ b/docs/reference/protocols/peer-agent-directory-and-observation-v0.md @@ -392,10 +392,11 @@ is omitted, and delivery remains available through the durable hand-off path. The first producer is `loopx agent-directory --goal-id [--agent-id ] [--cursor ]` (`loopx/cli_commands/agent_directory.py` over -`loopx/control_plane/agents/directory.py`). It reads the Goal's existing agent -management projection, so identity, work, claims and staleness keep their -current owners and the packet adds no second read of the registry or a lease -store. It is the same surface for both audiences: a peer Agent inside the Goal +`loopx/control_plane/agents/directory.py`, which supplies facts to the single +`agent.directory.project` TypeScript request). It consumes the Goal's existing +agent management projection and complete registration facts. Management work +selection and binding-route interpretation keep their existing owners; the +directory does not introduce a second lease store or routing resolver. It is the same surface for both audiences: a peer Agent inside the Goal and the steward channel call one command. What it emits, and what it refuses to imply: @@ -438,6 +439,42 @@ What it emits, and what it refuses to imply: It writes nothing. Reading it grants no claim, no lease, no priority and no work edit. +### Directory projection ownership and source evidence + +The internal request uses `peer_agent_directory_project_request_v0`; the public +packet remains `peer_agent_directory_v0`. +`loopx/control_plane/agents/directory.ts` owns directory membership checks, +work-field projection, route display limits, counts, continuation, limitations, +and the attention rollup. The rollup covers the returned page only; its category +counts include all matching page rows even when displayed items are capped at +eight. The Python adapter retains fact collection, the existing public-output +safety checks, transport and compatibility validation. A failed typed request +must not fall back to a Python implementation of directory decisions. + +The packet also carries the shared +[`projection_envelope`](../contracts/projection-envelope-contract.md). This is +a `loopx_projection_envelope_v0` envelope with projection `peer_agent_directory`, +an intentional additive field, not a change to `peer_agent_directory_v0`'s +existing row, scope, cursor or count semantics. The directory calls the existing +`sealProjectionEnvelope` owner in-process during its one projection request; +it does not add a separate sealing RPC or freshness algorithm. + +Upstream source rows and observation times survive composition. `collected_at` +is the directory assembly time, not proof that its sources were just read. +Serving or replaying a projection cannot refresh those source observations. +Stale, unreadable and incomplete upstream evidence remains visible; absent +upstream evidence is unknown, never freshly complete. A display page limit is +recorded separately from missing source coverage: undisplayed registered Agents +are not by themselves an unreadable source. Unavailable execution facts, an +observed empty fact set and partial Agent coverage remain distinct, and an +unknown lease must not become an absent lease. + +This read adds no presence, delivery, claim, lease or execution authority. No +persisted format or provider default changes. Retire the Python adapter when +the real directory callers and necessary fact providers can invoke the typed +owner directly; until then it is a compatibility boundary, not a second policy +owner. Rollback restores the prior code package without migrating business state. + ## Reference Implementation: Herdr [Herdr](https://github.com/herdrdev/herdr) is a terminal-space provider whose diff --git a/loopx/cli_commands/agent_directory.py b/loopx/cli_commands/agent_directory.py index dc196e8e56..e5c5f1e3ba 100644 --- a/loopx/cli_commands/agent_directory.py +++ b/loopx/cli_commands/agent_directory.py @@ -10,7 +10,8 @@ from ..agent_registry import load_goal_from_registry from ..control_plane.agents.directory import build_peer_agent_directory -from ..control_plane.effect_runtime import EffectRuntimeRemoteError +from ..control_plane.effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError +from ..control_plane.projection_envelope_facts import render_projection_envelope_markdown from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status @@ -68,7 +69,7 @@ def handle_agent_directory(args, registry_path, runtime_root, print_payload, out caller_agent_id=args.agent_id, cursor=args.cursor, ) - except (ValueError, TypeError, OSError, KeyError, EffectRuntimeRemoteError) as exc: + except (ValueError, TypeError, OSError, KeyError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: payload = { "ok": False, "schema_version": "peer_agent_directory_v0", @@ -131,6 +132,7 @@ def render_agent_directory(payload): presence = payload.get("presence_coverage") or {} lines.append(f"presence: {presence.get('state')} — {presence.get('note')}") lines.append("limitations: " + ", ".join(payload.get("limitations") or [])) + lines.extend(render_projection_envelope_markdown(payload.get("projection_envelope"))) lines.append( "Observation and delivery grant no claim, lease, priority or work edit." ) diff --git a/loopx/control_plane/agents/directory.py b/loopx/control_plane/agents/directory.py index 727c56a5a1..5f6fc03fbf 100644 --- a/loopx/control_plane/agents/directory.py +++ b/loopx/control_plane/agents/directory.py @@ -3,7 +3,8 @@ `peer_agent_directory_v0` (`docs/reference/protocols/peer-agent-directory-and-observation-v0.md`) answers what a peer Agent, or the steward Agent a person talks to, may see about the -other Agents of one Goal. This module is that contract's first local producer. +other Agents of one Goal. TypeScript owns the directory packet; this module +adapts its existing sources. It is a re-projection: the Agent rows come from the existing agent management projection, so identity, work, claims and staleness keep their current owners and this module adds no second read of the registry, the Todo index or a lease @@ -22,11 +23,12 @@ from __future__ import annotations -from collections.abc import Iterable, Mapping +from collections.abc import Mapping from typing import Any from ..runtime.public_safety import public_safe_compact_text from ..runtime.time import now_utc_iso +from ..effect_runtime import effect_runtime_result from ...thread_agent_binding import summarize_agent_binding_routes from ..runtime.public_safety import validate_public_safe_value from ..todos.contract import normalize_todo_id @@ -37,12 +39,8 @@ PEER_AGENT_DIRECTORY_SCHEMA_VERSION = "peer_agent_directory_v0" -PEER_AGENT_DIRECTORY_SCOPE = "goal_registered_agents" MAX_DIRECTORY_ROWS = 24 -MAX_OBSERVATION_REFS = 1 -MAX_DELIVERY_REFS = 1 MAX_ROUTE_CANDIDATES = 3 -MAX_ROLLUP_AGENTS = 8 # Limitation codes are contract vocabulary, not prose: a reader switches on them. LIMITATION_PRESENCE_PROVIDER_UNAVAILABLE = "presence_provider_unavailable" @@ -54,10 +52,6 @@ GAP_AUDIENCE_NOT_AUTHORIZED = "audience_not_authorized" -CLAIM_AGE_SUSPECTED_STALE = "suspected_stale" -CLAIM_AGE_FRESH = "fresh" -CLAIM_AGE_UNKNOWN = "unknown" - def _as_mapping(value: Any) -> Mapping[str, Any]: return value if isinstance(value, Mapping) else {} @@ -71,19 +65,17 @@ def _compact(value: Any, *, limit: int = 200) -> str | None: return public_safe_compact_text(value, limit=limit) -def _compact_refs(value: Any, *, limit: int) -> list[str]: +def _safe_refs(value: Any) -> list[str]: refs: list[str] = [] for item in _as_list(value): text = _compact(item, limit=160) - if text and text not in refs: + if text: refs.append(text) - if len(refs) >= limit: - break return refs -def _publishable_route(route: dict[str, Any]) -> tuple[dict[str, Any], int]: - """Cap the candidate list and withhold entries the public boundary rejects. +def _publishable_route(route: dict[str, Any]) -> dict[str, Any]: + """Withhold entries the existing public boundary rejects, without capping. Withholding is reported, never silent: `candidate_count` keeps the withheld entries, so a filtered candidate cannot be read as a binding that does not @@ -101,10 +93,10 @@ def _publishable_route(route: dict[str, Any]) -> tuple[dict[str, Any], int]: continue visible.append(candidate) published = {key: value for key, value in route.items() if key != "candidates"} - published["candidates"] = visible[:MAX_ROUTE_CANDIDATES] + published["candidates"] = visible if withheld: published["withheld_candidate_count"] = withheld - return published, withheld + return published def _projected_execution_facts(payload: Mapping[str, Any]) -> dict[str, Any] | None: @@ -129,71 +121,41 @@ def _projected_execution_facts(payload: Mapping[str, Any]) -> dict[str, Any] | N return facts -def _work_block(agent_row: Mapping[str, Any]) -> dict[str, Any] | None: - """Project one Agent's bounded work facts, or nothing when it holds none.""" +def _safe_agent_facts(row: Mapping[str, Any]) -> dict[str, Any]: + """Apply existing public safety, leaving directory rules to TypeScript. - todo = _as_mapping(agent_row.get("current_todo")) + Preserve fallback-before-compaction: a rejected preferred title must not + expose its fallback. This boundary supplies facts, not work classification. + """ + todo = _as_mapping(row.get("current_todo")) todo_id = _compact(todo.get("todo_id"), limit=80) - if not todo_id: - return None - claimed_by = _compact(todo.get("claimed_by"), limit=120) - stale_hint = _as_mapping(agent_row.get("stale_claim_hint")) - hint_state = _compact(stale_hint.get("state"), limit=60) - claim_age_state = ( - CLAIM_AGE_SUSPECTED_STALE - if hint_state == CLAIM_AGE_SUSPECTED_STALE - else CLAIM_AGE_FRESH - if claimed_by and _compact(todo.get("updated_at"), limit=60) - else CLAIM_AGE_UNKNOWN - ) - lease = _as_mapping(_as_mapping(agent_row.get("execution")).get("lease")) - work: dict[str, Any] = { - "todo_id": normalize_todo_id(todo_id) or todo_id, - "todo_status": _compact(todo.get("status"), limit=40) or "unknown", - "lease_status": _compact(lease.get("status"), limit=40), - "lease_expired": lease.get("expired") if isinstance(lease.get("expired"), bool) else None, - "task_class": _compact(todo.get("task_class"), limit=60), - "action_kind": _compact(todo.get("action_kind"), limit=60), - "priority": _compact(todo.get("priority"), limit=20), - "claimed": bool(claimed_by), - "claimed_by": claimed_by, - "claim_age_state": claim_age_state, + lease = _as_mapping(_as_mapping(row.get("execution")).get("lease")) + return { + "agent_id": _compact(row.get("agent_id"), limit=120), + "execution_fact_agent_id": str(row.get("agent_id") or "").strip(), + "agent_model": _compact(row.get("agent_model"), limit=60), + "current_todo": { + "todo_id": normalize_todo_id(todo_id) or todo_id, + "status": _compact(todo.get("status"), limit=40), + "task_class": _compact(todo.get("task_class"), limit=60), + "action_kind": _compact(todo.get("action_kind"), limit=60), + "priority": _compact(todo.get("priority"), limit=20), + "claimed_by": _compact(todo.get("claimed_by"), limit=120), + "updated_at": _compact(todo.get("updated_at"), limit=60), + "title": _compact(todo.get("title") or todo.get("text"), limit=220), + }, "last_activity_at": _compact( - agent_row.get("last_activity_at") or todo.get("updated_at"), limit=60 + row.get("last_activity_at") or todo.get("updated_at"), limit=60 ), - "title": _compact(todo.get("title") or todo.get("text"), limit=220), - } - return {key: value for key, value in work.items() if value is not None and value != ""} - - -def _rollup(rows: Iterable[Mapping[str, Any]]) -> dict[str, Any]: - """A typed "who needs a decision" ordering; it assigns nothing.""" - - needs_decision: list[str] = [] - stale_claims: list[str] = [] - unclaimed: list[str] = [] - for row in rows: - agent_id = _compact(row.get("agent_id"), limit=120) - if not agent_id: - continue - work = _as_mapping(row.get("work")) - if str(work.get("todo_status") or "") == "blocked": - needs_decision.append(agent_id) - if str(work.get("claim_age_state") or "") == CLAIM_AGE_SUSPECTED_STALE: - stale_claims.append(agent_id) - if not work.get("claimed"): - unclaimed.append(agent_id) - return { - "basis": "typed_work_state_only", - "assigns_work": False, - "needs_decision": needs_decision[:MAX_ROLLUP_AGENTS], - "stale_claims": stale_claims[:MAX_ROLLUP_AGENTS], - "without_claim": unclaimed[:MAX_ROLLUP_AGENTS], - "counts": { - "needs_decision": len(needs_decision), - "stale_claims": len(stale_claims), - "without_claim": len(unclaimed), + "stale_claim_hint": { + "state": _compact(_as_mapping(row.get("stale_claim_hint")).get("state"), limit=60), }, + "execution": {"lease": { + "status": _compact(lease.get("status"), limit=40), + "expired": lease.get("expired") if isinstance(lease.get("expired"), bool) else None, + }}, + "evidence_refs": _safe_refs(row.get("evidence_refs")), + "handoff_refs": _safe_refs(row.get("handoff_refs")), } @@ -206,37 +168,13 @@ def build_peer_agent_directory( execution_facts: Any = None, cursor: str | None = None, ) -> dict[str, Any]: - """Return a bounded `peer_agent_directory_v0` packet for one Goal. - - `caller_agent_id` is optional. When it is supplied the caller must be a - registered Agent of the Goal, since membership in this space is proven - against the registry rather than asserted by the caller; when it is absent - the packet records that the caller identity was not supplied instead of - inventing one. - - `execution_facts` defaults to the facts the payload's management projection - already carries on its rows, so this re-projection reads the same lane, - worker and lease facts status collection did instead of reading them again. - - `cursor` continues a previous page: rows follow the projection's agent-id - ordering and the cursor is the last id already read, so only Agents sorting - strictly after it are returned and `next_cursor` names the following page. - The default starts a fresh scan. Membership is proven by the Goal's - registration facts, never by the page and never by a Todo claim: paging - cannot deny a registered caller their scope, and a claim that outlived its - registration cannot admit a caller the Goal never registered. - - The packet's totals keep those two apart: `registered_agent_count` counts - the Goal's registrations and never a claimant, while `row_count` covers - every row published, including the diagnostic claimant rows a work - relationship adds (`registered: false`). `omitted_row_count` counts the rows - of that same agent-id order this packet did not carry, so a capped page - cannot read as the whole directory. - """ + """Collect existing facts and invoke the sole typed directory owner once. + Management selection and binding interpretation retain their owners. This + adapter normalizes inputs, applies public safety, and transports facts. + Failure never falls back to Python directory decisions. + """ payload = status_payload if isinstance(status_payload, Mapping) else {} - resolved_goal = _compact(goal_id or payload.get("goal_filter"), limit=120) - caller = _compact(caller_agent_id, limit=120) cursor_value = _compact(cursor, limit=120) if cursor and not cursor_value: raise ValueError("cursor is not a publishable agent id") @@ -251,147 +189,38 @@ def build_peer_agent_directory( after_agent_id=cursor_value, ) agent_rows = [row for row in _as_list(projection.get("agents")) if isinstance(row, Mapping)] - registered_ids = registered_agent_ids(status) - - limitations = [ - LIMITATION_PRESENCE_PROVIDER_UNAVAILABLE, - LIMITATION_PRESENCE_IS_ADVISORY, - ] - # The payload's published projection is itself bounded, so a page can hold - # Agents whose lane and lease evidence was never published. Declare that - # instead of letting those rows read as "no lease evidence". - facts_missing_for_page = facts is not None and any( - str(row.get("agent_id") or "").strip() not in facts for row in agent_rows - ) - if facts is None or facts_missing_for_page: - limitations.append(LIMITATION_LEASE_STATE_NOT_PROJECTED) - gaps: list[dict[str, Any]] = [] - if caller and caller not in registered_ids: - # An unregistered caller gets a scope gap, never a listing it has no - # scope over, and never a silent empty directory either. - gaps.append( - { - "kind": GAP_AUDIENCE_NOT_AUTHORIZED, - "detail": "caller_agent_id_is_not_registered_for_this_goal", - "caller_agent_id": caller, - } - ) - if not caller: - limitations.append(LIMITATION_CALLER_IDENTITY_NOT_SUPPLIED) - - # Route candidates are read from the same published bindings the management - # projection walks, but only the owning resolver module interprets them. binding_goals = _as_list(_as_mapping(payload.get("run_history")).get("goals")) - rows: list[dict[str, Any]] = [] - dropped_at_cap = 0 - withheld_route_candidates = 0 - for row in agent_rows: - agent_id = _compact(row.get("agent_id"), limit=120) - if not agent_id: - continue - if gaps: - break - if len(rows) >= MAX_DIRECTORY_ROWS: - dropped_at_cap += 1 - continue - work = _work_block(row) - peer_route, withheld_candidates = _publishable_route( - summarize_agent_binding_routes(binding_goals, agent_id=agent_id) + agents = [_safe_agent_facts(row) for row in agent_rows] + routes = { + row["agent_id"]: _publishable_route( + summarize_agent_binding_routes(binding_goals, agent_id=row["agent_id"]) ) - withheld_route_candidates += withheld_candidates - directory_row: dict[str, Any] = { - "agent_id": agent_id, - "registered": agent_id in registered_ids, - "agent_model": _compact(row.get("agent_model"), limit=60), - "work": work, - "observation_refs": _compact_refs( - row.get("evidence_refs"), limit=MAX_OBSERVATION_REFS - ), - "delivery_refs": _compact_refs( - row.get("handoff_refs"), limit=MAX_DELIVERY_REFS - ), - "peer_route": peer_route, - } - rows.append( - { - key: value - for key, value in directory_row.items() - if key in {"agent_id", "registered", "work"} or value not in (None, [], {}) - } - ) - # Registration facts are the membership total here too: this is the same - # set that decided the caller's scope and every row's `registered` flag, so - # a diagnostic claimant row can never inflate the Goal's registered count. - registered_count = len(registered_ids) - # The rows this scan reads are the Goal's whole directory: its registered - # Agents plus the diagnostic rows a work relationship adds. A cap applied - # upstream is reported as omitted rows rather than vanishing, so a reader - # cannot read a truncated page as "these are all the rows". - source_summary = _as_mapping(projection.get("source_summary")) - diagnostic_count = source_summary.get("diagnostic_agent_count") - diagnostic_count = diagnostic_count if isinstance(diagnostic_count, int) else 0 - source_count = registered_count + diagnostic_count - omitted = max(0, source_count - len(rows)) if not gaps else dropped_at_cap - if omitted: - limitations.append(LIMITATION_ROWS_TRUNCATED) - if withheld_route_candidates: - limitations.append(LIMITATION_ROUTE_CANDIDATE_WITHHELD) - - # A page can end before the registry does. `next_cursor` is the last id - # read, so the same agent-id ordering continues from exactly there instead - # of the 25th and later Agents becoming unreachable. - selected_count = source_summary.get("selected_agent_count") - more_rows_follow = False - if rows and not gaps: - if isinstance(selected_count, int): - more_rows_follow = selected_count > len(rows) - elif cursor_value is None: - more_rows_follow = source_count > len(rows) - next_cursor = str(rows[-1]["agent_id"]) if more_rows_follow else None - - packet: dict[str, Any] = { - "ok": True, - "schema_version": PEER_AGENT_DIRECTORY_SCHEMA_VERSION, - "goal_id": resolved_goal, - "collected_at": now_utc_iso(), - "scope": { - "mode": PEER_AGENT_DIRECTORY_SCOPE, - "caller_agent_id": caller, - "caller_membership": ( - "registered_agent" - if caller and caller in registered_ids - else "unregistered_agent" - if caller - else "local_surface" - ), - "scope_basis": "goal_registry", - "gaps": gaps, - }, - "rows": rows, - "row_count": len(rows), - "omitted_row_count": omitted, - "registered_agent_count": registered_count, - "cursor": cursor_value, - "next_cursor": next_cursor, - "presence_coverage": { - "provider": None, - "state": "unavailable", - "note": ( - "presence answers 'runnable now'; no presence provider is " - "registered for this Goal, so rows carry registry identity and " - "durable work state only" - ), - }, - "authority": { - "observation_grants": [], - "writes": False, - "scheduler": False, - "note": "observation and delivery grant no claim, lease, priority or work edit", - }, - "limitations": limitations, + for row in agents if row["agent_id"] } - if diagnostic_count: - packet["diagnostic_agent_count"] = diagnostic_count - if rows: - packet["rollup"] = _rollup(rows) - return packet + upstream_envelopes: list[Mapping[str, Any]] = [] + for source in (payload, _as_mapping(payload.get("agent_management_projection")), projection): + envelope = source.get("projection_envelope") + if isinstance(envelope, Mapping) and envelope not in upstream_envelopes: + upstream_envelopes.append(envelope) + result = effect_runtime_result("agent.directory.project", { + "schema_version": "peer_agent_directory_project_request_v0", + "observed_at": now_utc_iso(), + "goal_id": _compact(goal_id or payload.get("goal_filter"), limit=120), + "caller_agent_id": _compact(caller_agent_id, limit=120), + "cursor": cursor_value, + "registered_agent_ids": sorted(registered_agent_ids(status)), + "agents": agents, + "source_summary": dict(_as_mapping(projection.get("source_summary"))), + "execution_fact_agent_ids": list(facts) if facts is not None else None, + "routes": routes, + "upstream_envelopes": upstream_envelopes, + }) + if ( + not isinstance(result, dict) + or result.get("schema_version") != PEER_AGENT_DIRECTORY_SCHEMA_VERSION + or result.get("ok") is not True + or not isinstance(result.get("rows"), list) + or not isinstance(result.get("projection_envelope"), dict) + ): + raise TypeError("invalid peer agent directory projection result") + return result diff --git a/loopx/control_plane/agents/directory.ts b/loopx/control_plane/agents/directory.ts new file mode 100644 index 0000000000..a263eb00fd --- /dev/null +++ b/loopx/control_plane/agents/directory.ts @@ -0,0 +1,142 @@ +/** Peer directory decision owner. IO, public-safety filtering and management + * projection stay with their existing adapters; this projection reads no clock. */ +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { jsonObject, requireJsonObject, requireNonEmptyString, requireStringArray, + optionalNonEmptyString } from "../runtime_decode.ts"; +import { sealProjectionEnvelope, PROJECTION_ENVELOPE_SEAL_REQUEST } from "../projection_envelope.ts"; + +type ObjectRow = Record; +const object = (value: unknown): ObjectRow => jsonObject(value) ?? {}; +const text = (value: unknown): string | null => typeof value === "string" && value.length ? value : null; +const integer = (value: unknown): number | null => typeof value === "number" && Number.isInteger(value) ? value : null; +function array(value: unknown, label: string): unknown[] { + if (!Array.isArray(value)) throw new EffectRuntimeRequestError(`${label} must be an array`); + return value; +} +function validateAgent(row: ObjectRow, index: number): ObjectRow { + const label = `agents[${index}]`; + const strings = (input: ObjectRow, keys: string[], prefix: string) => { + for (const key of keys) { + const value = input[key]; + if (value !== undefined && value !== null && typeof value !== "string") { + throw new EffectRuntimeRequestError(`${prefix}.${key} must be a string or null`); + } + } + }; + strings(row, ["agent_id", "agent_model", "execution_fact_agent_id", "last_activity_at"], label); + for (const key of ["evidence_refs", "handoff_refs"]) { + if (row[key] !== undefined) requireStringArray(row[key], `${label}.${key}`); + } + if (row.current_todo !== undefined) strings(requireJsonObject(row.current_todo, `${label}.current_todo`), + ["todo_id", "status", "task_class", "action_kind", "priority", "claimed_by", "updated_at", "title"], `${label}.current_todo`); + if (row.stale_claim_hint !== undefined) strings(requireJsonObject(row.stale_claim_hint, `${label}.stale_claim_hint`), ["state"], `${label}.stale_claim_hint`); + if (row.execution !== undefined) { + const execution = requireJsonObject(row.execution, `${label}.execution`); + if (execution.lease !== undefined) { + const lease = requireJsonObject(execution.lease, `${label}.execution.lease`); + strings(lease, ["status"], `${label}.execution.lease`); + if (lease.expired !== undefined && lease.expired !== null && typeof lease.expired !== "boolean") { + throw new EffectRuntimeRequestError(`${label}.execution.lease.expired must be a boolean or null`); + } + } + } + return row; +} +function refs(value: unknown): string[] { + return [...new Set(Array.isArray(value) ? value.filter((v): v is string => typeof v === "string" && !!v) : [])].slice(0, 1); +} +function workBlock(row: ObjectRow): ObjectRow | null { + const todo = object(row.current_todo), hint = object(row.stale_claim_hint); + if (!text(todo.todo_id)) return null; + const claimedBy = text(todo.claimed_by), lease = object(object(row.execution).lease); + const work: ObjectRow = { + todo_id: todo.todo_id, todo_status: text(todo.status) ?? "unknown", + lease_status: text(lease.status), lease_expired: typeof lease.expired === "boolean" ? lease.expired : null, + task_class: text(todo.task_class), action_kind: text(todo.action_kind), priority: text(todo.priority), + claimed: !!claimedBy, claimed_by: claimedBy, + claim_age_state: hint.state === "suspected_stale" ? "suspected_stale" : claimedBy && text(todo.updated_at) ? "fresh" : "unknown", + last_activity_at: text(row.last_activity_at), title: text(todo.title), + }; + return Object.fromEntries(Object.entries(work).filter(([, value]) => value !== null && value !== "")); +} +function rollup(rows: ObjectRow[]): ObjectRow { + const needs = rows.filter(row => object(row.work).todo_status === "blocked").map(row => row.agent_id); + const stale = rows.filter(row => object(row.work).claim_age_state === "suspected_stale").map(row => row.agent_id); + const unclaimed = rows.filter(row => !object(row.work).claimed).map(row => row.agent_id); + return {basis: "typed_work_state_only", assigns_work: false, needs_decision: needs.slice(0, 8), + stale_claims: stale.slice(0, 8), without_claim: unclaimed.slice(0, 8), + counts: {needs_decision: needs.length, stale_claims: stale.length, without_claim: unclaimed.length}}; +} + +export function projectPeerAgentDirectory(value: unknown): ObjectRow { + const request = requireJsonObject(value, "peer directory request"); + if (request.schema_version !== "peer_agent_directory_project_request_v0") { + throw new EffectRuntimeRequestError("peer directory request schema mismatch"); + } + const now = requireNonEmptyString(request.observed_at, "observed_at"); + const caller = optionalNonEmptyString(request.caller_agent_id, "caller_agent_id"); + const goal = optionalNonEmptyString(request.goal_id, "goal_id"); + const cursor = optionalNonEmptyString(request.cursor, "cursor"); + const registered = new Set(requireStringArray(request.registered_agent_ids, "registered_agent_ids")); + const agents = array(request.agents, "agents").map((row, index) => validateAgent(requireJsonObject(row, `agents[${index}]`), index)); + const summary = requireJsonObject(request.source_summary, "source_summary"); + const routes = requireJsonObject(request.routes, "routes"); + const executionIds = request.execution_fact_agent_ids === null ? null + : new Set(requireStringArray(request.execution_fact_agent_ids, "execution_fact_agent_ids")); + const upstream = array(request.upstream_envelopes, "upstream_envelopes"); + // Validate the route facts before projecting so malformed requests cannot be + // silently accepted just because the caller or display cap hides a row. + for (const [id, value] of Object.entries(routes)) { + const route = requireJsonObject(value, `routes.${id}`); + array(route.candidates, `routes.${id}.candidates`).forEach((candidate, index) => requireJsonObject(candidate, `routes.${id}.candidates[${index}]`)); + if (route.withheld_candidate_count !== undefined && (integer(route.withheld_candidate_count) === null || Number(route.withheld_candidate_count) < 0)) { + throw new EffectRuntimeRequestError(`routes.${id}.withheld_candidate_count must be a non-negative integer`); + } + } + const limitations = ["presence_provider_unavailable", "presence_is_advisory"]; + if (executionIds === null || agents.some(row => !executionIds.has(String(row.execution_fact_agent_id ?? row.agent_id ?? "").trim()))) { + limitations.push("lease_state_not_projected"); + } + const gaps: ObjectRow[] = caller && !registered.has(caller) ? [{kind: "audience_not_authorized", + detail: "caller_agent_id_is_not_registered_for_this_goal", caller_agent_id: caller}] : []; + if (!caller) limitations.push("caller_identity_not_supplied"); + const rows: ObjectRow[] = []; + let withheld = 0; + for (const row of agents) { + const id = text(row.agent_id); + if (!id) continue; + if (gaps.length || rows.length >= 24) break; + const route = requireJsonObject(routes[id], `routes.${id}`); + withheld += Number(route.withheld_candidate_count ?? 0); + const projected: ObjectRow = {agent_id: id, registered: registered.has(id), work: workBlock(row), + peer_route: {...route, candidates: (route.candidates as unknown[]).slice(0, 3)}}; + if (text(row.agent_model)) projected.agent_model = row.agent_model; + const observations = refs(row.evidence_refs), deliveries = refs(row.handoff_refs); + if (observations.length) projected.observation_refs = observations; + if (deliveries.length) projected.delivery_refs = deliveries; + rows.push(projected); + } + const diagnostic = integer(summary.diagnostic_agent_count) ?? 0; + const sourceCount = registered.size + diagnostic; + const omitted = gaps.length ? 0 : Math.max(0, sourceCount - rows.length); + if (omitted) limitations.push("rows_truncated_at_cap"); + if (withheld) limitations.push("route_candidate_withheld"); + const selected = integer(summary.selected_agent_count); + const more = rows.length > 0 && !gaps.length && (selected !== null ? selected > rows.length : cursor === null && sourceCount > rows.length); + const packet: ObjectRow = {ok: true, schema_version: "peer_agent_directory_v0", goal_id: goal, collected_at: now, + scope: {mode: "goal_registered_agents", caller_agent_id: caller, + caller_membership: caller ? registered.has(caller) ? "registered_agent" : "unregistered_agent" : "local_surface", + scope_basis: "goal_registry", gaps}, rows, row_count: rows.length, omitted_row_count: omitted, + registered_agent_count: registered.size, cursor, next_cursor: more ? rows.at(-1)!.agent_id : null, + presence_coverage: {provider: null, state: "unavailable", note: "presence answers 'runnable now'; no presence provider is registered for this Goal, so rows carry registry identity and durable work state only"}, + authority: {observation_grants: [], writes: false, scheduler: false, note: "observation and delivery grant no claim, lease, priority or work edit"}, limitations}; + if (diagnostic) packet.diagnostic_agent_count = diagnostic; + if (rows.length) packet.rollup = rollup(rows); + packet.projection_envelope = sealProjectionEnvelope({schema_version: PROJECTION_ENVELOPE_SEAL_REQUEST, + projection: "peer_agent_directory", observed_at: now, sources: upstream.length ? [] : [{ + source_id: "directory_inputs", read_status: "not_read", required: true, last_read_at: null}], upstream, + coverage: {scope: "goal_registered_agents", expected_count: upstream.length ? sourceCount : null, + included_count: gaps.length ? 0 : sourceCount, shown_count: rows.length, available_count: sourceCount, + omitted: gaps.length && sourceCount ? [{reason: "audience_not_authorized", count: sourceCount}] : []}}); + return packet; +} diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 85ec873139..8731e3e980 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -275,6 +275,7 @@ export function createEffectRuntimeHandlers( ["capabilities.periodic_report.approval_retry.select", lazyHandler(() => import("./capabilities/periodic_report_progress.ts"), ({selectPeriodicReportApprovalRetry}) => selectPeriodicReportApprovalRetry)], ["todo.succession.project", lazyHandler(() => import("./todos/succession.ts"), ({projectTodoSuccession}) => projectTodoSuccession)], ["todo.work_counts.project", lazyHandler(() => import("./todos/summary_lanes.ts"), ({projectLegacyTodoWorkCounts}) => projectLegacyTodoWorkCounts)], + ["agent.directory.project", lazyHandler(() => import("./agents/directory.ts"), ({projectPeerAgentDirectory}) => projectPeerAgentDirectory)], ["projection.envelope.seal", lazyHandler(() => import("./projection_envelope.ts"), ({sealProjectionEnvelope}) => sealProjectionEnvelope)], ["todo.global_readiness.project", lazyHandler(() => import("./todos/decision_scope.ts"), ({projectGlobalTodoReadiness}) => projectGlobalTodoReadiness)], ["todo.decision_scope.evaluate", lazyHandler(() => import("./todos/decision_scope.ts"), ({evaluateDecisionScope}) => evaluateDecisionScope)], diff --git a/tests/control_plane/test_peer_agent_directory.py b/tests/control_plane/test_peer_agent_directory.py index 34eb0f3e90..0b5e03cda5 100644 --- a/tests/control_plane/test_peer_agent_directory.py +++ b/tests/control_plane/test_peer_agent_directory.py @@ -490,7 +490,8 @@ def test_a_cursor_that_cannot_be_published_is_rejected_not_echoed() -> None: ) -def test_cli_agent_directory_continues_with_cursor(tmp_path) -> None: +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_cli_agent_directory_continues_with_cursor(tmp_path, provider) -> None: registered = [f"agent-{index:02d}" for index in range(MAX_DIRECTORY_ROWS + 4)] registry = tmp_path / "registry.json" registry.write_text( @@ -514,13 +515,14 @@ def test_cli_agent_directory_continues_with_cursor(tmp_path) -> None: config["common_runtime_root"] = str(runtime) config["goals"][0].update(repo=str(tmp_path), state_file=str(state)) registry.write_text(json.dumps(config), encoding="utf-8") - initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state, provider="sqlite") + initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state, provider=provider) - def run(*extra: str) -> dict[str, object]: + def run(*extra: str, runtime_override: str | None = None) -> dict[str, object]: output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = cli_main( [ + *(["--runtime-root", runtime_override] if runtime_override else []), "--registry", str(registry), "--format", @@ -536,6 +538,7 @@ def run(*extra: str) -> dict[str, object]: assert exit_code == 0 return json.loads(output.getvalue()) + before = state.read_bytes(), registry.read_bytes() first = run() assert first["row_count"] == MAX_DIRECTORY_ROWS assert first["next_cursor"] @@ -547,6 +550,12 @@ def run(*extra: str) -> dict[str, object]: assert sorted(ids) == registered assert len(ids) == len(set(ids)) assert second["next_cursor"] is None + assert (state.read_bytes(), registry.read_bytes()) == before + explicit = run("--cursor", str(first["next_cursor"]), runtime_override=str(runtime)) + assert explicit["rows"] == second["rows"] + denied = run("--agent-id", "agent-not-registered") + assert denied["rows"] == [] + assert denied["scope"]["gaps"][0]["kind"] == GAP_AUDIENCE_NOT_AUTHORIZED def test_rows_report_a_peer_route_without_selecting_one() -> None: @@ -680,3 +689,85 @@ def test_unavailable_source_does_not_resurrect_project_asset_or_hint() -> None: packet = build_peer_agent_directory(payload) assert _rows_by_agent(packet)[WORKING_AGENT]["work"] is None assert projected_agent_goals(payload)[GOAL_ID][WORKING_AGENT]["open_todo_ids"] == [] + + +def test_adapter_sends_one_projection_request_with_complete_membership(monkeypatch) -> None: + from loopx.control_plane.agents import directory + + ids = [f"agent-{index:02d}" for index in range(MAX_DIRECTORY_ROWS + 3)] + payload = _status_payload(ids) + source = {"schema_version": "synthetic-envelope"} + payload["projection_envelope"] = source + calls = [] + result = {"ok": True, "schema_version": PEER_AGENT_DIRECTORY_SCHEMA_VERSION, + "rows": [], "projection_envelope": {}} + + def project(method, request): + calls.append((method, request)) + return result + + monkeypatch.setattr(directory, "effect_runtime_result", project) + monkeypatch.setattr(directory, "now_utc_iso", lambda: "2026-01-02T00:00:00Z") + assert build_peer_agent_directory(payload, execution_facts={}) is result + assert len(calls) == 1 + method, request = calls[0] + assert method == "agent.directory.project" + assert request["registered_agent_ids"] == ids + assert len(request["agents"]) == MAX_DIRECTORY_ROWS + assert request["execution_fact_agent_ids"] == [] + assert request["upstream_envelopes"] == [source] + assert request["observed_at"] == "2026-01-02T00:00:00Z" + + +@pytest.mark.parametrize("result", [None, {}, {"schema_version": PEER_AGENT_DIRECTORY_SCHEMA_VERSION}]) +def test_adapter_rejects_malformed_result_without_python_fallback(monkeypatch, result) -> None: + from loopx.control_plane.agents import directory + + monkeypatch.setattr(directory, "effect_runtime_result", lambda *args: result) + with pytest.raises(TypeError, match="invalid peer agent directory projection result"): + build_peer_agent_directory(_status_payload()) + + +def test_adapter_propagates_failed_projection_without_fallback(monkeypatch) -> None: + from loopx.control_plane.agents import directory + + def fail(*args): + raise OSError("typed runtime unavailable") + + monkeypatch.setattr(directory, "effect_runtime_result", fail) + with pytest.raises(OSError, match="typed runtime unavailable"): + build_peer_agent_directory(_status_payload()) + + +def test_missing_source_envelope_is_unknown_and_visible() -> None: + packet = build_peer_agent_directory(_status_payload()) + envelope = packet["projection_envelope"] + assert envelope["fresh"] is False + assert envelope["complete"] is False + assert envelope["sources"][0]["last_read_at"] is None + assert "projection alerts" in render_agent_directory(packet) + + +def test_adapter_preserves_safe_text_fallback_and_unicode_boundaries(monkeypatch) -> None: + from loopx.control_plane.agents import directory + + row = { + "agent_id": WORKING_AGENT, + "last_activity_at": "/tmp/private-activity", + "current_todo": { + "todo_id": "todo_directory_text", + "title": "/tmp/private-title", + "text": "This fallback must not replace a rejected preferred title", + "updated_at": "2026-01-01T00:00:00Z", + "claimed_by": WORKING_AGENT, + }, + } + monkeypatch.setattr(directory, "build_agent_management_projection", lambda *args, **kwargs: { + "agents": [row], "source_summary": {"selected_agent_count": 1}, + }) + work = build_peer_agent_directory(_status_payload())["rows"][0]["work"] + assert "title" not in work + assert "last_activity_at" not in work + row["current_todo"]["title"] = "🌱" * 225 + work = build_peer_agent_directory(_status_payload())["rows"][0]["work"] + assert work["title"] == "🌱" * 219 + "…" diff --git a/tests/control_plane_ts/peer_agent_directory.test.ts b/tests/control_plane_ts/peer_agent_directory.test.ts new file mode 100644 index 0000000000..2cbc772279 --- /dev/null +++ b/tests/control_plane_ts/peer_agent_directory.test.ts @@ -0,0 +1,101 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {projectPeerAgentDirectory} from "../../loopx/control_plane/agents/directory.ts"; +import {sealProjectionEnvelope, PROJECTION_ENVELOPE_SEAL_REQUEST} from "../../loopx/control_plane/projection_envelope.ts"; +const now = "2026-01-01T01:00:00Z"; +type Row = Record; +function request(count = 1): Row { + const ids = Array.from({length: count}, (_, i) => `agent-${String(i).padStart(2, "0")}`); + return {schema_version: "peer_agent_directory_project_request_v0", observed_at: now, goal_id: "goal", + caller_agent_id: null, cursor: null, registered_agent_ids: ids, agents: ids.map(agent_id => ({agent_id})), + source_summary: {selected_agent_count: count}, execution_fact_agent_ids: null, + routes: Object.fromEntries(ids.map(id => [id, {state: "unbound", candidate_count: 0, candidates: []}])), upstream_envelopes: []}; +} +function source(read_status = "read", included_count = 1): Row { + return sealProjectionEnvelope({schema_version: PROJECTION_ENVELOPE_SEAL_REQUEST, projection: "status", + observed_at: "2026-01-01T00:00:00Z", sources: [{source_id: "registry", read_status, + last_read_at: "2026-01-01T00:00:00Z", source_updated_at: "2025-12-31T23:00:00Z"}], + coverage: {scope: "goal", expected_count: 1, included_count}}); +} +test("membership uses whole registry beyond page and diagnostic claimants never gain scope", () => { + const input = request(30); input.agents = input.agents.slice(0, 24); input.caller_agent_id = "agent-29"; + const result: Row = projectPeerAgentDirectory(input); + assert.equal(result.scope.caller_membership, "registered_agent"); + assert.equal(result.rows.length, 24); assert.equal(result.next_cursor, "agent-23"); + assert.equal(result.omitted_row_count, 6); + input.caller_agent_id = "claimant"; + assert.equal((projectPeerAgentDirectory(input) as Row).rows.length, 0); + assert.equal((projectPeerAgentDirectory(input) as Row).scope.gaps[0].kind, "audience_not_authorized"); +}); +test("later and exhausted pages retain total omission definition", () => { + const input = request(30); input.agents = input.agents.slice(24); input.cursor = "agent-23"; + input.source_summary.selected_agent_count = 6; + let result: Row = projectPeerAgentDirectory(input); + assert.equal(result.omitted_row_count, 24); assert.equal(result.next_cursor, null); + input.agents = []; input.source_summary.selected_agent_count = 0; + result = projectPeerAgentDirectory(input); assert.equal(result.omitted_row_count, 30); +}); +test("null work, false flags, lease coverage and bounded references retain semantics", () => { + const input = request(2); input.registered_agent_ids = ["agent-00"]; + input.source_summary.diagnostic_agent_count = 1; + input.agents[0] = {agent_id: "agent-00", current_todo: {todo_id: "T1", status: "blocked", title: "Task"}, + execution: {lease: {expired: false}}, evidence_refs: ["a", "a", "b"], handoff_refs: ["c", "d"]}; + input.execution_fact_agent_ids = ["agent-00"]; + const result: Row = projectPeerAgentDirectory(input); + assert.equal(result.rows[0].work.lease_expired, false); assert.equal(result.rows[0].work.claimed, false); + assert.equal(result.rows[1].registered, false); assert.equal(result.rows[1].work, null); + assert.deepEqual(result.rows[0].observation_refs, ["a"]); assert.deepEqual(result.rows[0].delivery_refs, ["c"]); + assert.ok(result.limitations.includes("lease_state_not_projected")); + input.execution_fact_agent_ids = ["agent-00", "agent-01"]; + assert.ok(!(projectPeerAgentDirectory(input) as Row).limitations.includes("lease_state_not_projected")); + input.execution_fact_agent_ids = []; + assert.ok((projectPeerAgentDirectory(input) as Row).limitations.includes("lease_state_not_projected")); +}); +test("route cap does not erase withheld candidates or full rollup counts", () => { + const input = request(12); + for (const row of input.agents) row.current_todo = {todo_id: "T1", status: "blocked"}; + input.routes["agent-00"] = {state: "ambiguous", candidate_count: 7, withheld_candidate_count: 2, + candidates: Array.from({length: 5}, (_, i) => ({thread_id: `t${i}`}))}; + const result: Row = projectPeerAgentDirectory(input); + assert.equal(result.rows[0].peer_route.candidates.length, 3); assert.equal(result.rows[0].peer_route.candidate_count, 7); + assert.equal(result.rollup.needs_decision.length, 8); assert.equal(result.rollup.counts.needs_decision, 12); + assert.ok(result.limitations.includes("route_candidate_withheld")); +}); +test("missing evidence is unknown, and replay preserves old source timestamps", () => { + const input = request(); + let result: Row = projectPeerAgentDirectory(input); + assert.equal(result.projection_envelope.fresh, false); assert.equal(result.projection_envelope.complete, false); + assert.equal(result.projection_envelope.coverage.expected_count, null); + input.upstream_envelopes = [source()]; + result = projectPeerAgentDirectory(input); + assert.equal(result.projection_envelope.sources[0].last_read_at, "2026-01-01T00:00:00Z"); + assert.equal(result.projection_envelope.sources[0].source_updated_at, "2025-12-31T23:00:00Z"); + assert.equal(result.projection_envelope.fresh, false); assert.equal(result.projection_envelope.served_at, now); + assert.ok(result.projection_envelope.alert_reasons.includes("stale_sources")); + input.upstream_envelopes = [source("unreadable")]; + assert.ok((projectPeerAgentDirectory(input) as Row).projection_envelope.alert_reasons.includes("unreadable_sources")); + input.upstream_envelopes = [source("read", 0)]; + assert.equal((projectPeerAgentDirectory(input) as Row).projection_envelope.complete, false); +}); +test("display paging is not missing source coverage", () => { + const input = request(30); input.upstream_envelopes = [source()]; + const result: Row = projectPeerAgentDirectory(input); + assert.equal(result.projection_envelope.complete, true); + assert.equal(result.projection_envelope.coverage.truncated, true); + assert.deepEqual(result.projection_envelope.coverage.omitted, []); +}); +test("strict bridge shape rejects malformed input without mutation", () => { + for (const patch of [{schema_version: "old"}, {agents: {}}, {registered_agent_ids: [2]}, + {execution_fact_agent_ids: false}, {routes: []}, {observed_at: "not-time"}, + {upstream_envelopes: [{}]}]) assert.throws(() => projectPeerAgentDirectory({...request(), ...patch})); + const input = request(); const before = structuredClone(input); + projectPeerAgentDirectory(input); assert.deepEqual(input, before); +}); +test("runtime dispatch reaches the single directory owner and validates nested facts", async () => { + const {createEffectRuntimeHandlers, dispatchEffectRuntimeMethod} = await import("../../loopx/control_plane/effect_runtime_handlers.ts"); + const handlers = createEffectRuntimeHandlers({fingerprint: "test", requestShutdown: () => undefined}); + assert.deepEqual(await dispatchEffectRuntimeMethod(handlers, "agent.directory.project", request()), projectPeerAgentDirectory(request())); + for (const row of [{agent_id: 2}, {current_todo: []}, {evidence_refs: [false]}, {execution: {lease: {expired: "false"}}}]) { + await assert.rejects(dispatchEffectRuntimeMethod(handlers, "agent.directory.project", {...request(), agents: [row]})); + } +}); diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index b211b87e2b..c24cf9bba5 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -116,6 +116,7 @@ "tests/control_plane_ts/goal_instance_identity.test.ts", "tests/control_plane_ts/chat_session_lifecycle.test.ts", "tests/control_plane_ts/test_python_runtime.test.ts", + "tests/control_plane_ts/peer_agent_directory.test.ts", "tests/control_plane_ts/direction_authority.test.ts", "tests/control_plane_ts/direction_baseline.test.ts", "tests/control_plane_ts/zcode*.test.ts" From 983caa3230d097ff54623b511e525835769688cf Mon Sep 17 00:00:00 2001 From: aihop <631479666@qq.com> Date: Fri, 6 Nov 2026 08:19:17 +0800 Subject: [PATCH 2/2] docs(agents): record peer directory qualification evidence --- .../2026-11-06-peer-directory.md | 74 ++++++++++++++----- .../2026-11-06-peer-directory.zh-CN.md | 59 +++++++++++---- 2 files changed, 101 insertions(+), 32 deletions(-) diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md index 5e8e7f397d..917d963bc3 100644 --- a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md +++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.md @@ -44,24 +44,62 @@ CLI a one-request operation. Rule ownership and retained source evidence are the intended benefit; moving code alone is not evidence of a speedup or better whole-Goal outcomes. -Qualification is pending until recorded against the final integrated candidate: - -- Fixed-time baseline/head comparison through the actual command, preserving - old fields, ordering, diagnostics and exit status, with envelope additions - checked separately. -- File and SQLite first/last pages, later-page members, unauthorized claimants, - explicit runtime roots and unchanged business state. -- Stale, unreadable, incomplete, missing and replayed source evidence; - independent mutations must detect page-only membership and refreshed old - source timestamps. -- Clean wheel and sdist execution without borrowing the source checkout. -- Matched whole-CLI cold/warm p50/p95, runtime request counts and output sizes; - type checking, native/Python regressions and required premerge checks. - -This checklist is not a passed-test claim. Preserve failures and unmeasured -boundaries when adding results; earlier CLI optimization measurements do not -qualify this candidate. This checkpoint does not close T4 or change release -qualification. +## Measured candidate qualification + +Candidate `9d91c5bc178a0acddbbab3276b404b9d0ceada3f` was checked against the +fixed baseline above. The results below qualify this bounded synthetic fixture, +not a universal performance guarantee or completion of T4: + +- Native TypeScript tests: 6,129 passed, 77 skipped; focused Python tests: + 33 passed. Randomized comparison covered 160 parity cases. Both independent + mutations (page-only membership and refreshing old observation timestamps) + were caught. +- Eight actual CLI baseline/head pairs covered File and SQLite first page, + last page, cursor past the end and unregistered caller, including a registered + caller beyond the first page. Existing fields, order, diagnostics and exit + status matched; the new envelope was checked separately. Registry, Goal, + canonical authority and fence data stayed unchanged. Initial SQLite reads + created empty WAL/SHM sidecars, so this is not a zero-filesystem-write claim. +- Fresh wheel and sdist environments passed 16 installed console commands + across both providers and the four cases, outside the checkout with installed + import provenance and matching runtime-source hashes. Minimum-Node and + Windows qualification were not performed. +- Premerge exited 0 with an inherited advisory. Skips and the advisory remain + limitations, not additional passed checks. + +### Whole-command cost + +Linux, Python 3.12.14 and Node 24.22.1 measurements used identical synthetic +28-Agent canonical File/SQLite fixtures with an empty work set, explicit runtime +roots and usage telemetry disabled. The final alternating baseline/head run +contained 120 command executions: 20 warm and 10 cold samples per provider and +version. Warm means a new Python process using a running TS server; cold means +a new Python process and isolated runtime directory without a TS server. OS +filesystem caches were not cleared. Times include the complete subprocess; +p95 uses nearest rank on these small samples. + +| Provider/runtime | Baseline p50 / p95 (ms) | Candidate p50 / p95 (ms) | +| --- | --- | --- | +| File warm | 348.85 / 397.95 | 348.14 / 380.69 | +| File cold | 949.17 / 986.70 | 950.08 / 1026.45 | +| SQLite warm | 342.77 / 396.21 | 338.94 / 373.52 | +| SQLite cold | 947.77 / 1020.42 | 953.57 / 1078.73 | + +Every measured command went from five to six runtime requests. First-page +stdout grew from 9,330 to 11,384 UTF-8 bytes (+2,054; 22.0%). Cold p95 rose +39.75 ms (4.0%) for File and 58.31 ms (5.7%) for SQLite; lower warm samples +do not establish a speedup or statistical significance. This records the +additional crossing and envelope cost, not a latency or release qualification. +Earlier CLI optimization results do not qualify this candidate. + +The first paired attempt reused long-idle runtime directories and hit a SQLite +transport failure in the unchanged `explore.configuration.goal` call before +directory execution (exit 1). The underlying cause is not established. That +attempt's results remain separate negative evidence. A complete rerun with fresh +isolated runtime directories passed all 120 measurements with unchanged fixture +hashes; it was not a selective retry of the failing sample. Long-idle runtime +reliability remains unqualified. This checkpoint does not close T4 or change +release qualification. ## Bridge exit and rollback diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md index fb5fab4fdd..8dd1c4e559 100644 --- a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md +++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-06-peer-directory.zh-CN.md @@ -34,20 +34,51 @@ Python 本地工作，本次为该阶段增加一次粗粒度运行时请求； 只有一次请求**。收益是目录规则收口及保留来源依据，代码迁移本身不能证明加速 或整 Goal 成果改善。 -以下资格待最终集成候选的实测结果记录后确认： - -- 实际命令的固定时间 baseline/head 差分，保持旧字段、顺序、诊断及退出状态， - 单独核对 envelope 增量。 -- File／SQLite 首末页、后页成员、未授权 claimant、显式 runtime root，以及 - 业务状态不变。 -- 过期、不可读、不完整、缺失及重放来源依据；独立变异必须能检出只按当前页 - 判断成员以及用当前时间刷新旧来源时间的错误。 -- 不借用源码 checkout 的干净 wheel／sdist 安装执行。 -- 同负载完整 CLI 冷／暖 p50、p95、运行时请求次数及输出体积；类型检查、 - 原生／Python 回归及必需的 premerge 检查。 - -此清单不表示测试已通过。补充结果时须保留失败和未测边界；既往 CLI 优化测量 -不能资格化本候选。本检查点不关闭 T4，也不改变发布资格。 +## 候选实测资格 + +候选 `9d91c5bc178a0acddbbab3276b404b9d0ceada3f` 与上述固定基线比较。 +以下结果只覆盖本次有界合成 fixture，不代表通用性能保证或 T4 完成： + +- 原生 TypeScript 测试 6,129 项通过、77 项跳过；focused Python 测试 33 项 + 通过。随机差分覆盖 160 个等价案例；仅以当前页判断成员、以当前时间刷新旧 + 观察时间这两种独立变异均被检出。 +- 八组实际 CLI baseline/head 配对覆盖 File／SQLite 首页、末页、越界 cursor + 和未注册调用方，包括位于首页之外的注册调用方。旧字段、顺序、诊断及退出状态 + 一致，新增 envelope 单独核对。registry、Goal、canonical authority 和 fence + 数据不变；SQLite 初次读取创建了空 WAL／SHM sidecar，因此不宣称零文件系统写入。 +- 独立 wheel／sdist 环境在 checkout 外通过 16 条安装态 console 命令，覆盖 + 两个 provider 和四种案例；确认导入自安装目录且运行时源码 hash 一致。 + 未执行最低 Node 版本或 Windows 资格验证。 +- Premerge 退出码为 0，并保留一项继承的 advisory。跳过项与 advisory 仍是 + 验收限制，不能算成额外通过的检查。 + +### 完整命令成本 + +测量环境为 Linux、Python 3.12.14、Node 24.22.1，使用相同的合成 28-Agent +canonical File／SQLite fixture，工作集为空，显式指定 runtime root 并关闭 +usage telemetry。最终交替 baseline/head 的测量共 120 次命令执行：每个 +provider、每个版本各 20 次暖态和 10 次冷态。暖态为新 Python 进程复用运行中 +TS server；冷态为新 Python 进程及无 TS server 的隔离 runtime 目录；未清除 +操作系统文件缓存。计时包含整个子进程，小样本 p95 使用 nearest-rank 方法。 + +| Provider／runtime | 基线 p50／p95（ms） | 候选 p50／p95（ms） | +| --- | --- | --- | +| File 暖态 | 348.85 / 397.95 | 348.14 / 380.69 | +| File 冷态 | 949.17 / 986.70 | 950.08 / 1026.45 | +| SQLite 暖态 | 342.77 / 396.21 | 338.94 / 373.52 | +| SQLite 冷态 | 947.77 / 1020.42 | 953.57 / 1078.73 | + +每次测量的运行时请求数均从五次增至六次。首页 stdout 从 9,330 增至 11,384 +UTF-8 字节（增加 2,054，22.0%）。File 冷态 p95 增加 39.75 ms（4.0%）， +SQLite 增加 58.31 ms（5.7%）；较低的暖态样本不能证明加速或统计显著性。 +本记录量化新增 crossing 和 envelope 的成本，不授予延迟或发布资格；此前的 +CLI 优化结果不能资格化本候选。 + +首轮配对尝试复用了长时间空闲的 runtime 目录，在目录执行前、未改动的 +`explore.configuration.goal` 调用中遇到一次 SQLite transport 失败，退出码为 1； +确切根因尚未确认。该轮结果作为独立负面证据保留。随后使用全新隔离 runtime +目录完整重跑，120 次测量全部通过且 fixture hash 不变，并非仅重试失败样本。 +长时间空闲 runtime 的可靠性仍未获资格确认。本检查点不关闭 T4，也不改变发布资格。 ## Bridge 退出与回滚