fix(multi-subagent): distinguish configured and live child capacity

docs/integrations/codex-subagent-orchestration.md

@@ -522,6 +522,36 @@ loopx agent-context --goal-id example-peer-task-goal --agent-id coordinator \
   --phase after_delegate_result --format json
 ```
 
+`max_children` is the configured Goal ceiling, not a live count of available
+native host slots. When a native `spawn` or `followup` call reports the bounded
+`agent_thread_limit_reached` outcome, return that typed observation without raw
+host text:
+
+```bash
+loopx agent-context --goal-id example-peer-task-goal --agent-id coordinator \
+  --phase after_delegate_result --native-child-operation spawn \
+  --native-child-outcome agent_thread_limit_reached --native-child-count 1 \
+  --format json
+```
+
+The resulting `native_host_capacity` fact is read-only and scoped to the current
+host observation. It tells the coordinator to stop same-Turn spawn/followup
+retries, mark unlaunched work incomplete and continue useful parent work until
+capacity changes. It does not delete, import, resume or rebind sessions, lower
+the configured ceiling, or claim that a completed child freed a slot. Raw host
+errors and session identities are never accepted by this interface. A
+`succeeded` observation proves only that operation; it does not authorize
+another same-Turn attempt or claim that additional capacity remains.
+
+中文：`max_children` 只是 Goal 配置上限，不代表宿主此刻有同样数量的可用槽位。
+当原生 `spawn` 或 `followup` 返回 `agent_thread_limit_reached` 时，使用上述
+`after_delegate_result` 调用提交有界类型化观察；返回的
+`native_host_capacity` 会要求本 Turn 停止重复派发、把未启动工作标记为未完成，
+并继续主 Agent 的有用工作，待容量变化后再试。该只读接口不会删除、导入、恢复或
+重新绑定任何 Session，也不会把“子任务已完成”臆断成“容量已经释放”。一次
+`succeeded` 观察只证明该次操作成功，不会授权本 Turn 再次尝试，也不表示仍有
+额外容量。
+
 The coordinator must be registered. The command reads current registry policy
 without writing a Todo, starting a turn or spending quota. When an execution
 configuration is present, `before_plan` reads only the binding directory;

loopx/cli_commands/agent_context.py

@@ -16,6 +16,21 @@ def register_agent_context(subparsers, add_format):
         required=True,
         choices=("before_plan", "before_delegate", "after_delegate_result"),
     )
+    parser.add_argument(
+        "--native-child-operation",
+        choices=("spawn", "followup"),
+        help="Typed native child operation observed by the current host.",
+    )
+    parser.add_argument(
+        "--native-child-outcome",
+        choices=("succeeded", "agent_thread_limit_reached"),
+        help="Typed native child outcome; raw host errors are never accepted.",
+    )
+    parser.add_argument(
+        "--native-child-count",
+        type=int,
+        help="Optional non-negative native child count observed by the host.",
+    )
 
 
 def handle_agent_context(args, registry_path, runtime_root, print_payload, output_format):
@@ -27,25 +42,89 @@ def handle_agent_context(args, registry_path, runtime_root, print_payload, outpu
             render_agent_context,
         )
         return 1
+    operation = args.native_child_operation
+    outcome = args.native_child_outcome
+    child_count = args.native_child_count
+    if bool(operation) != bool(outcome):
+        print_payload(
+            {
+                "ok": False,
+                "error": (
+                    "--native-child-operation and --native-child-outcome "
+                    "must be provided together"
+                ),
+            },
+            output_format(args),
+            render_agent_context,
+        )
+        return 1
+    if operation and args.phase != "after_delegate_result":
+        print_payload(
+            {
+                "ok": False,
+                "error": "native child outcomes require --phase after_delegate_result",
+            },
+            output_format(args),
+            render_agent_context,
+        )
+        return 1
+    if child_count is not None and (child_count < 0 or not operation):
+        print_payload(
+            {
+                "ok": False,
+                "error": (
+                    "--native-child-count must be non-negative and accompany "
+                    "a native child operation/outcome"
+                ),
+            },
+            output_format(args),
+            render_agent_context,
+        )
+        return 1
+    observations = {}
+    if operation:
+        native_capacity = {
+            "schema_version": "native_subagent_capacity_observation_v0",
+            "operation": operation,
+            "outcome": outcome,
+        }
+        if child_count is not None:
+            native_capacity["child_count"] = child_count
+        observations["native_host_capacity"] = native_capacity
     context = project_goal_agent_context(
         phase=args.phase,
         scope={"goal_id": args.goal_id, "agent_id": args.agent_id, "todo_id": None},
         goal=goal,
         registry_path=registry_path,
         runtime_root=runtime_root,
+        observations=observations,
     )
-    print_payload(
-        {
-            "ok": True,
-            "agent_context": context,
-            "source": "registry.spawn_policy+local_delegation",
-            "read_only": True,
-            "host_receipts_observed": False,
-            "host_receipts_scope": "native_tool_input",
-        },
-        output_format(args),
-        render_agent_context,
-    )
+    if operation and context is None:
+        print_payload(
+            {
+                "ok": False,
+                "error": "native child outcomes require enabled multi_subagent policy",
+            },
+            output_format(args),
+            render_agent_context,
+        )
+        return 1
+    payload = {
+        "ok": True,
+        "agent_context": context,
+        "source": "registry.spawn_policy+local_delegation",
+        "read_only": True,
+        "host_receipts_observed": False,
+        "host_receipts_scope": "native_tool_input",
+    }
+    if operation:
+        payload.update(
+            {
+                "host_capacity_observed": True,
+                "host_capacity_scope": "native_tool_input",
+            }
+        )
+    print_payload(payload, output_format(args), render_agent_context)
     return 0
 
 

loopx/control_plane/subagent_context.ts

@@ -4,12 +4,12 @@ import type { JsonObject } from "./effect_program.ts";
 import { jsonObject, requireJsonObject } from "./runtime_decode.ts";
 
 export const subagentContextProvider: AgentContextProvider = {
-  hookId: "multi_subagent.coordinator", capabilityId: "multi_subagent", revision: "v3",
+  hookId: "multi_subagent.coordinator", capabilityId: "multi_subagent", revision: "v4",
   phases: AGENT_CONTEXT_PHASES,
   produce(input, config) {
     const guidance = {
       before_plan: [
-        "Prefer parallel delegation for bounded independent work when useful; avoid duplicate reads. Keep one decision-relevant coordinator question.",
+        "Prefer bounded independent delegation; max_children is a configured ceiling, not live availability. Admit native children incrementally, avoid duplicate reads, and keep one parent question.",
         "Native child tools can read loopx agent-context at before_delegate and after_delegate_result; these calls do not start Turns or spend quota.",
         "Authorized routes are observations, not obligations. Use a ready route only when it fits; blocked/unknown routes never block native work. No heartbeat must use every route.",
       ],
@@ -19,11 +19,17 @@ export const subagentContextProvider: AgentContextProvider = {
       ],
       after_delegate_result: [
         "Check returned sources, omissions and contradictions against the question. Reconcile native child receipts and any freshly read bound delegation operation receipts; missing, unavailable or rejected receipts do not establish completed work.",
+        "On typed agent_thread_limit_reached, stop same-Turn spawn/followup retries, mark unlaunched work incomplete, and continue useful parent work.",
         "Verify decisive sources and record accept/defer/reject with reasons. Link accepted evidence to the deliverable and run parent validation before writeback; opinions are not independent evidence.",
       ],
     }[input.phase];
     const facts: JsonObject = {
       max_children: config.max_children,
+      capacity_contract: {
+        schema_version: "multi_subagent_capacity_v0",
+        configured_limit_kind: "upper_bound",
+        live_availability: "not_observed",
+      },
       model_preference: jsonObject(config.model_config),
     };
     const count = input.observations.child_count;
@@ -33,6 +39,15 @@ export const subagentContextProvider: AgentContextProvider = {
       facts.delegation_context = delegation;
     }
     if (input.phase === "after_delegate_result") {
+      const nativeCapacity = boundedNativeCapacityObservation(
+        input.observations.native_host_capacity,
+      );
+      if (nativeCapacity) {
+        facts.native_host_capacity = nativeCapacity;
+        const contract = facts.capacity_contract as JsonObject;
+        contract.live_availability = nativeCapacity.outcome === "agent_thread_limit_reached"
+          ? "capacity_exhausted" : "attempt_observed";
+      }
       const counts = jsonObject(input.observations.reconciliation_counts);
       facts.receipt_observation = counts ? "host_reconciled" : "not_supplied";
       if (counts) facts.reconciliation_counts = Object.fromEntries(
@@ -51,6 +66,38 @@ export const subagentContextProvider: AgentContextProvider = {
   },
 };
 
+function boundedNativeCapacityObservation(value: unknown): JsonObject | null {
+  const source = jsonObject(value);
+  if (!source || source.schema_version !== "native_subagent_capacity_observation_v0") {
+    return null;
+  }
+  const operation = String(source.operation ?? "");
+  const outcome = String(source.outcome ?? "");
+  if (!["spawn", "followup"].includes(operation)
+    || !["succeeded", "agent_thread_limit_reached"].includes(outcome)) {
+    return null;
+  }
+  const result: JsonObject = {
+    schema_version: "native_subagent_capacity_observation_v0",
+    operation,
+    outcome,
+  };
+  const childCount = source.child_count;
+  if (Number.isInteger(childCount) && Number(childCount) >= 0) {
+    result.child_count = Math.min(Number(childCount), 10_000);
+  }
+  if (outcome === "agent_thread_limit_reached") {
+    result.retry_same_turn = false;
+    result.reason_code = "agent_thread_limit_reached";
+    result.recovery_actions = [
+      "continue_parent_work",
+      "defer_unlaunched_children",
+      "retry_after_capacity_change",
+    ];
+  }
+  return result;
+}
+
 function boundedDelegationContext(value: unknown): JsonObject | null {
   const source = jsonObject(value);
   if (!source || source.schema_version !== "loopx_delegation_context_v0") return null;

tests/control_plane/test_agent_context.py

@@ -283,8 +283,89 @@ def test_native_cli_is_read_only_and_does_not_claim_native_receipts(tmp_path, ph
     assert payload["agent_context"]["phase"] == phase
     assert payload["host_receipts_observed"] is False
     assert payload["host_receipts_scope"] == "native_tool_input"
+    assert "host_capacity_observed" not in payload
+    assert "host_capacity_scope" not in payload
     assert registry.read_bytes() == before
     command[command.index(SCOPE["agent_id"])] = "unregistered"
     rejected = subprocess.run(command, capture_output=True, text=True)
     assert rejected.returncode == 1
     assert json.loads(rejected.stdout)["ok"] is False
+
+
+def test_native_cli_projects_typed_capacity_exhaustion_without_raw_error(tmp_path):
+    registry = tmp_path / "registry.json"
+    registry.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "goals": [
+                    {
+                        "id": SCOPE["goal_id"],
+                        "repo": str(tmp_path),
+                        "status": "active",
+                        "registered_agents": [SCOPE["agent_id"]],
+                        "spawn_policy": POLICY,
+                    }
+                ],
+            }
+        )
+    )
+    before = registry.read_bytes()
+    command = [
+        sys.executable,
+        "-m",
+        "loopx.cli",
+        "--registry",
+        str(registry),
+        "agent-context",
+        "--goal-id",
+        SCOPE["goal_id"],
+        "--agent-id",
+        SCOPE["agent_id"],
+        "--phase",
+        "after_delegate_result",
+        "--native-child-operation",
+        "spawn",
+        "--native-child-outcome",
+        "agent_thread_limit_reached",
+        "--native-child-count",
+        "1",
+        "--format",
+        "json",
+    ]
+
+    result = subprocess.run(command, capture_output=True, text=True, check=True)
+    payload = json.loads(result.stdout)
+    [contribution] = payload["agent_context"]["contributions"]
+    observation = contribution["facts"]["native_host_capacity"]
+    assert payload["host_capacity_observed"] is True
+    assert payload["host_receipts_observed"] is False
+    assert observation["reason_code"] == "agent_thread_limit_reached"
+    assert observation["retry_same_turn"] is False
+    assert observation["recovery_actions"] == [
+        "continue_parent_work",
+        "defer_unlaunched_children",
+        "retry_after_capacity_change",
+    ]
+    assert registry.read_bytes() == before
+
+    invalid = subprocess.run(
+        [
+            *command[: command.index("after_delegate_result")],
+            "before_plan",
+            *command[command.index("after_delegate_result") + 1 :],
+        ],
+        capture_output=True,
+        text=True,
+    )
+    assert invalid.returncode == 1
+    assert "after_delegate_result" in json.loads(invalid.stdout)["error"]
+
+    disabled_payload = json.loads(registry.read_text())
+    disabled_payload["goals"][0]["spawn_policy"]["spawn_allowed"] = False
+    registry.write_text(json.dumps(disabled_payload))
+    disabled_before = registry.read_bytes()
+    disabled = subprocess.run(command, capture_output=True, text=True)
+    assert disabled.returncode == 1
+    assert "enabled multi_subagent" in json.loads(disabled.stdout)["error"]
+    assert registry.read_bytes() == disabled_before

tests/control_plane/test_turn_envelope_budget_warning.py

@@ -158,7 +158,13 @@ def test_installed_skill_defers_delegation_policy_to_enabled_provider(tmp_path):
     enabled = project_agent_context(
         phase="before_plan", scope=SCOPE, orchestration=POLICY
     )
+    contribution = enabled["contributions"][0]
     assert any(
-        "parallel delegation" in item
-        for item in enabled["contributions"][0]["guidance"]
+        "max_children is a configured ceiling" in item
+        for item in contribution["guidance"]
     )
+    assert contribution["facts"]["capacity_contract"] == {
+        "schema_version": "multi_subagent_capacity_v0",
+        "configured_limit_kind": "upper_bound",
+        "live_availability": "not_observed",
+    }