From d3465b169c669524eb83df79b763bd3c1b673ba9 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sun, 20 Sep 2026 08:44:13 +0800 Subject: [PATCH 1/4] fix(multi-subagent): project native host capacity Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../codex-subagent-orchestration.md | 26 +++++++ loopx/cli_commands/agent_context.py | 67 ++++++++++++++++++ loopx/control_plane/subagent_context.ts | 51 +++++++++++++- tests/control_plane/test_agent_context.py | 70 +++++++++++++++++++ tests/control_plane_ts/agent_context.test.ts | 41 ++++++++++- 5 files changed, 252 insertions(+), 3 deletions(-) diff --git a/docs/integrations/codex-subagent-orchestration.md b/docs/integrations/codex-subagent-orchestration.md index 908171527..2c70c229f 100644 --- a/docs/integrations/codex-subagent-orchestration.md +++ b/docs/integrations/codex-subagent-orchestration.md @@ -522,6 +522,32 @@ 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. + +中文：`max_children` 只是 Goal 配置上限，不代表宿主此刻有同样数量的可用槽位。 +当原生 `spawn` 或 `followup` 返回 `agent_thread_limit_reached` 时，使用上述 +`after_delegate_result` 调用提交有界类型化观察；返回的 +`native_host_capacity` 会要求本 Turn 停止重复派发、把未启动工作标记为未完成， +并继续主 Agent 的有用工作，待容量变化后再试。该只读接口不会删除、导入、恢复或 +重新绑定任何 Session，也不会把“子任务已完成”臆断成“容量已经释放”。 + 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; diff --git a/loopx/cli_commands/agent_context.py b/loopx/cli_commands/agent_context.py index 4b1a9766e..5e6769495 100644 --- a/loopx/cli_commands/agent_context.py +++ b/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,12 +42,62 @@ 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( { @@ -42,6 +107,8 @@ def handle_agent_context(args, registry_path, runtime_root, print_payload, outpu "read_only": True, "host_receipts_observed": False, "host_receipts_scope": "native_tool_input", + "host_capacity_observed": bool(operation), + "host_capacity_scope": "native_tool_input", }, output_format(args), render_agent_context, diff --git a/loopx/control_plane/subagent_context.ts b/loopx/control_plane/subagent_context.ts index 2e3aadf4a..5670d0a0f 100644 --- a/loopx/control_plane/subagent_context.ts +++ b/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, + retry_same_turn: outcome !== "agent_thread_limit_reached", + }; + 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.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; diff --git a/tests/control_plane/test_agent_context.py b/tests/control_plane/test_agent_context.py index 0c7db75c0..a09ab65a1 100644 --- a/tests/control_plane/test_agent_context.py +++ b/tests/control_plane/test_agent_context.py @@ -288,3 +288,73 @@ def test_native_cli_is_read_only_and_does_not_claim_native_receipts(tmp_path, ph 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"] diff --git a/tests/control_plane_ts/agent_context.test.ts b/tests/control_plane_ts/agent_context.test.ts index 772deda10..696e58ac7 100644 --- a/tests/control_plane_ts/agent_context.test.ts +++ b/tests/control_plane_ts/agent_context.test.ts @@ -188,9 +188,48 @@ test("coordinator participation guidance survives all bounded lifecycle projecti } })!; assert.deepEqual(packet.failures, []); const [contribution] = packet.contributions as Record[]; - assert.equal(contribution.revision, "v3"); + assert.equal(contribution.revision, "v4"); assert.equal(packet.authority, "guidance_only"); assert.ok(Buffer.byteLength(JSON.stringify(contribution)) <= 2048); assert.ok(Buffer.byteLength(JSON.stringify(packet)) <= 3072); } }); + +test("configured child limit stays distinct from typed native host capacity", () => { + const before = evaluateSubagentContext({ phase: "before_plan", scope, + orchestration: { ...policy, max_children: 6 } })!; + const beforeFacts = (before.contributions as Record[])[0].facts; + assert.equal(beforeFacts.max_children, 6); + assert.deepEqual(beforeFacts.capacity_contract, { + schema_version: "multi_subagent_capacity_v0", + configured_limit_kind: "upper_bound", + live_availability: "not_observed", + }); + + const after = evaluateSubagentContext({ phase: "after_delegate_result", scope, + orchestration: { ...policy, max_children: 6 }, observations: { + native_host_capacity: { + schema_version: "native_subagent_capacity_observation_v0", + operation: "followup", + outcome: "agent_thread_limit_reached", + child_count: 1, + raw_error: "private host detail", + }, + } })!; + const afterFacts = (after.contributions as Record[])[0].facts; + assert.equal(afterFacts.capacity_contract.live_availability, "capacity_exhausted"); + assert.deepEqual(afterFacts.native_host_capacity, { + schema_version: "native_subagent_capacity_observation_v0", + operation: "followup", + outcome: "agent_thread_limit_reached", + retry_same_turn: false, + child_count: 1, + reason_code: "agent_thread_limit_reached", + recovery_actions: [ + "continue_parent_work", + "defer_unlaunched_children", + "retry_after_capacity_change", + ], + }); + assert.ok(!JSON.stringify(after).includes("private host detail")); +}); From c326d0cc2cca2007ca8823475a2b7d512d7985db Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sun, 20 Sep 2026 17:59:34 +0800 Subject: [PATCH 2/4] test(multi-subagent): assert typed capacity semantics Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../control_plane/test_turn_envelope_budget_warning.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/control_plane/test_turn_envelope_budget_warning.py b/tests/control_plane/test_turn_envelope_budget_warning.py index f1557ea51..c96f671d5 100644 --- a/tests/control_plane/test_turn_envelope_budget_warning.py +++ b/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", + } From d41a7087f477ae09e399c16e157aabf14b14dfb7 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sun, 20 Sep 2026 20:06:48 +0800 Subject: [PATCH 3/4] fix(multi-subagent): keep successful capacity unknown Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- docs/integrations/codex-subagent-orchestration.md | 8 ++++++-- loopx/control_plane/subagent_context.ts | 2 +- tests/control_plane_ts/agent_context.test.ts | 12 ++++++++++++ 3 files changed, 19 insertions(+), 3 deletions(-) diff --git a/docs/integrations/codex-subagent-orchestration.md b/docs/integrations/codex-subagent-orchestration.md index 2c70c229f..b758c1ae8 100644 --- a/docs/integrations/codex-subagent-orchestration.md +++ b/docs/integrations/codex-subagent-orchestration.md @@ -539,14 +539,18 @@ 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. +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，也不会把“子任务已完成”臆断成“容量已经释放”。 +重新绑定任何 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 diff --git a/loopx/control_plane/subagent_context.ts b/loopx/control_plane/subagent_context.ts index 5670d0a0f..785249aa3 100644 --- a/loopx/control_plane/subagent_context.ts +++ b/loopx/control_plane/subagent_context.ts @@ -81,13 +81,13 @@ function boundedNativeCapacityObservation(value: unknown): JsonObject | null { schema_version: "native_subagent_capacity_observation_v0", operation, outcome, - retry_same_turn: outcome !== "agent_thread_limit_reached", }; 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", diff --git a/tests/control_plane_ts/agent_context.test.ts b/tests/control_plane_ts/agent_context.test.ts index 696e58ac7..db8ac5437 100644 --- a/tests/control_plane_ts/agent_context.test.ts +++ b/tests/control_plane_ts/agent_context.test.ts @@ -232,4 +232,16 @@ test("configured child limit stays distinct from typed native host capacity", () ], }); assert.ok(!JSON.stringify(after).includes("private host detail")); + + const succeeded = evaluateSubagentContext({ phase: "after_delegate_result", scope, + orchestration: { ...policy, max_children: 6 }, observations: { + native_host_capacity: { + schema_version: "native_subagent_capacity_observation_v0", + operation: "spawn", + outcome: "succeeded", + }, + } })!; + const succeededFacts = (succeeded.contributions as Record[])[0].facts; + assert.equal(succeededFacts.capacity_contract.live_availability, "attempt_observed"); + assert.equal(succeededFacts.native_host_capacity.retry_same_turn, undefined); }); From 2acb4f86428efa38d967c60578a955db75932ef4 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sun, 20 Sep 2026 20:14:02 +0800 Subject: [PATCH 4/4] fix(multi-subagent): isolate native capacity input Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/cli_commands/agent_context.py | 40 +++++++++++++++-------- tests/control_plane/test_agent_context.py | 11 +++++++ 2 files changed, 37 insertions(+), 14 deletions(-) diff --git a/loopx/cli_commands/agent_context.py b/loopx/cli_commands/agent_context.py index 5e6769495..9ce1658ff 100644 --- a/loopx/cli_commands/agent_context.py +++ b/loopx/cli_commands/agent_context.py @@ -99,20 +99,32 @@ def handle_agent_context(args, registry_path, runtime_root, print_payload, outpu 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", - "host_capacity_observed": bool(operation), - "host_capacity_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 diff --git a/tests/control_plane/test_agent_context.py b/tests/control_plane/test_agent_context.py index a09ab65a1..c14a29b00 100644 --- a/tests/control_plane/test_agent_context.py +++ b/tests/control_plane/test_agent_context.py @@ -283,6 +283,8 @@ 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) @@ -358,3 +360,12 @@ def test_native_cli_projects_typed_capacity_exhaustion_without_raw_error(tmp_pat ) 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