feat(agents): ship the first local producer of the peer agent directory

docs/architecture/rfcs/harness-selection-dsh-pi-v0.md

@@ -608,6 +608,11 @@ State Providers](./shared-goal-authority-state-provider-v0.md).
   limits: observation and delivery grant nothing, a bounded wait pins the
   identity it resolved and requires the observed state to move, and a "who needs
   a decision now" rollup orders attention without assigning work.
+  The contract's first local producer ships as `loopx agent-directory --goal-id
+  <goal> [--agent-id <caller>]`, which re-projects the existing agent management
+  projection instead of reading the registry twice, reports no presence while no
+  presence provider is registered, and returns a typed scope gap instead of rows
+  to a caller that is not a registered Agent of that Goal.
 - **The authoritative per-lane readback is the alignment projection.** Once a
   plan lands, a lane's state is what `shared_goal_alignment_v0` reports for that
   Agent (`loopx shared-goal-alignment --goal-id <goal> --agent-id <agent>`):

docs/architecture/rfcs/harness-selection-dsh-pi-v0.zh-CN.md

@@ -468,6 +468,10 @@ Todo 创建、quota 或 goal policy——复用预览点名的身份，不得扩
   出一个管家专用 directory。两类受众遵守同样的边界：观察与投递不授予任何东西；一次有界
   等待会 pin 它已解析的身份、并要求观察到的状态确实向前变过；"现在谁需要决策"的 rollup
   只排序注意力，不分配任何工作。
+  该契约的首个本地 producer 已出货：`loopx agent-directory --goal-id <goal>
+  [--agent-id <caller>]`，它复用既有的 agent management projection 而不第二次读取
+  registry；在没有 presence provider 注册时不报告 presence；调用方若不是该 Goal 的已注册
+  Agent，得到的是 typed scope gap 而不是行。
 - **每条 lane 的权威回读是对齐投影。** 计划落地后，一条 lane 的状态就是该 Agent 的
   `shared_goal_alignment_v0` 投影（`loopx shared-goal-alignment --goal-id <goal> --agent-id
   <agent>`）：规范修订、frontier basis、claim 与租约事实、可领取的未认领工作。apply 的回执

docs/reference/protocols/peer-agent-directory-and-observation-v0.md

@@ -279,6 +279,34 @@ LoopX ships no requirement that a provider exists. With no provider, the
 directory degenerates to registered identity plus durable work state, presence
 is omitted, and delivery remains available through the durable hand-off path.
 
+### Local producer (shipped)
+
+The first producer is `loopx agent-directory --goal-id <goal> [--agent-id
+<caller>]` (`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
+and the steward channel call one command.
+
+What it emits, and what it refuses to imply:
+
+- one row per registered Agent, whether or not that Agent holds projected work,
+  and no `presence` block at all while no provider is registered;
+- explicit `limitations` (`presence_provider_unavailable`,
+  `presence_is_advisory`, `lease_state_not_projected`,
+  `caller_identity_not_supplied`, `rows_truncated_at_cap`) together with
+  `registered_agent_count` and `omitted_row_count`, so a truncated directory
+  cannot be read as a complete one;
+- a typed scope gap (`audience_not_authorized`) and zero rows when the named
+  caller is not a registered Agent of the Goal, instead of a listing that caller
+  has no scope over;
+- a typed-only rollup that orders attention by projected work state and assigns
+  nothing.
+
+It writes nothing. Reading it grants no claim, no lease, no priority and no work
+edit.
+
 ## Reference Implementation: Herdr
 
 [Herdr](https://github.com/herdrdev/herdr) is a terminal-space provider whose

loopx/capabilities/manager_context/skills/loopx-manager/SKILL.md

@@ -55,6 +55,16 @@ For a routine report or priority question, focus on active Goals. Do not
 inspect stopped Goals just to fill a report. The filter uses Core activation
 state, never age, stale progress, missing evidence or lack of recent activity.
 
+For "who else is working on this Goal", or which peer needs a decision, the
+in-space read is the peer directory: `loopx agent-directory --goal-id <goal>
+--agent-id <your own agent id>`. It lists each registered Agent of that Goal
+with the work it currently holds. Read its `limitations` before answering: while
+no presence provider is registered the packet carries registry identity and
+durable work state only, so it cannot say whether a peer is running right now,
+and a caller that is not a registered Agent of that Goal receives a scope gap
+instead of rows. Reading the directory grants no claim, lease or priority over
+the work it shows.
+
 For an all-Goal report, inspect relevant Goals and dates, then synthesize their
 concrete results. For "what needs me", read current owner tasks and explain
 the decision, consequence and work that can continue. Group related findings;

loopx/cli.py

@@ -4,6 +4,7 @@
 import sys
 
 from .cli_commands.agent_capabilities import register_agent_capabilities, handle_agent_capabilities
+from .cli_commands.agent_directory import register_agent_directory, handle_agent_directory
 from .cli_commands.agent_context import register_agent_context, handle_agent_context
 from .cli_commands.todo_continuation import register_todo_continuation, handle_todo_continuation
 from .cli_commands.manager_inbox import register_manager_inbox, handle_manager_inbox
@@ -329,6 +330,7 @@ def build_parser() -> LoopXArgumentParser:
     register_manager_inbox(sub, add_subcommand_format)
     register_agent_capabilities(sub, add_subcommand_format)
     register_agent_context(sub, add_subcommand_format)
+    register_agent_directory(sub, add_subcommand_format)
     register_lark_inbox_commands(sub, add_subcommand_format)
     register_lark_kanban_commands(sub, add_subcommand_format)
 
@@ -768,6 +770,12 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "agent-context":
         return handle_agent_context(args, registry_path, print_payload, output_format)
 
+    if args.command == "agent-directory":
+        return handle_agent_directory(
+            args, registry_path, effective_runtime_root(registry_path, args.runtime_root),
+            print_payload, output_format,
+        )
+
     if args.command == "manager-inbox":
         return handle_manager_inbox(args, registry_path, effective_runtime_root(registry_path, args.runtime_root))
 

loopx/cli_commands/agent_directory.py

@@ -0,0 +1,119 @@
+"""Read the Agent-facing peer directory of one Goal (read-only).
+
+`peer_agent_directory_v0` is the contract for one Agent discovering, observing
+and delivering to another. This command is its local producer: the steward
+channel and a peer Agent inside the Goal call the same surface, and the packet
+they get names its own limits instead of implying presence it cannot see.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+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 ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status
+
+
+def register_agent_directory(subparsers, add_format):
+    parser = subparsers.add_parser(
+        "agent-directory",
+        help="Read the peer agent directory of one Goal (read-only).",
+    )
+    add_format(parser)
+    parser.add_argument("--goal-id", required=True)
+    parser.add_argument(
+        "--agent-id",
+        default=None,
+        help=(
+            "The calling Agent. Supplying it proves membership against the Goal "
+            "registry; an unregistered caller receives a scope gap instead of rows."
+        ),
+    )
+    parser.add_argument(
+        "--scan-path",
+        action="append",
+        default=[],
+        help="Project root to scan for Goal state; defaults to the current directory.",
+    )
+
+
+def handle_agent_directory(args, registry_path, runtime_root, print_payload, output_format):
+    goal = load_goal_from_registry(registry_path, args.goal_id)
+    if not goal:
+        payload = {
+            "ok": False,
+            "schema_version": "peer_agent_directory_v0",
+            "error_code": "goal_not_registered",
+            "goal_id": args.goal_id,
+            "reason": (
+                "the registry does not know this Goal, so there is no Agent "
+                "identity to report and no scope to read"
+            ),
+        }
+        print_payload(payload, output_format(args), render_agent_directory)
+        return 1
+    try:
+        scan_roots = [Path(item).expanduser() for item in args.scan_path] or [Path.cwd()]
+        status_payload = collect_status(
+            registry_path=registry_path,
+            runtime_root_override=str(runtime_root) if runtime_root else args.runtime_root,
+            scan_roots=scan_roots,
+            limit=AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK,
+            goal_id=args.goal_id,
+        )
+        payload = build_peer_agent_directory(
+            status_payload,
+            goal_id=args.goal_id,
+            caller_agent_id=args.agent_id,
+        )
+    except (ValueError, TypeError, OSError, KeyError, EffectRuntimeRemoteError) as exc:
+        payload = {
+            "ok": False,
+            "schema_version": "peer_agent_directory_v0",
+            "error_code": "agent_directory_collection_failed",
+            "goal_id": args.goal_id,
+            "reason": str(exc),
+        }
+    print_payload(payload, output_format(args), render_agent_directory)
+    return 0 if payload.get("ok") else 1
+
+
+def render_agent_directory(payload):
+    if not payload.get("ok"):
+        return f"Peer agent directory unavailable: {payload.get('error_code')} — {payload.get('reason')}"
+    lines = [
+        f"Peer agent directory: {payload.get('goal_id')} "
+        f"({payload.get('row_count')} of {payload.get('row_count', 0) + payload.get('omitted_row_count', 0)} registered Agents)"
+    ]
+    scope = payload.get("scope") or {}
+    lines.append(
+        f"caller: {scope.get('caller_agent_id') or '(not supplied)'} "
+        f"({scope.get('caller_membership')})"
+    )
+    for gap in scope.get("gaps") or []:
+        lines.append(f"scope gap: {gap.get('kind')} — {gap.get('detail')}")
+    for row in payload.get("rows") or []:
+        work = row.get("work") or {}
+        lines.append(
+            f"- {row.get('agent_id')} [{row.get('agent_model') or 'unknown'}] "
+            f"{work.get('todo_id') or 'no projected work'} "
+            f"{work.get('todo_status') or ''} {work.get('priority') or ''} "
+            f"claim_age={work.get('claim_age_state') or 'n/a'}"
+        )
+    rollup = payload.get("rollup")
+    if rollup:
+        lines.append(
+            "rollup (typed attention only; assigns nothing): "
+            f"needs_decision={rollup['counts']['needs_decision']} "
+            f"stale_claims={rollup['counts']['stale_claims']} "
+            f"without_claim={rollup['counts']['without_claim']}"
+        )
+    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.append(
+        "Observation and delivery grant no claim, lease, priority or work edit."
+    )
+    return "\n".join(lines)