diff --git a/docs/integrations/codex-subagent-orchestration.md b/docs/integrations/codex-subagent-orchestration.md index 908171527..b758c1ae8 100644 --- a/docs/integrations/codex-subagent-orchestration.md +++ b/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; diff --git a/loopx/cli_commands/agent_context.py b/loopx/cli_commands/agent_context.py index 4b1a9766e..9ce1658ff 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,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 diff --git a/loopx/control_plane/subagent_context.ts b/loopx/control_plane/subagent_context.ts index 2e3aadf4a..785249aa3 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, + }; + 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; diff --git a/tests/control_plane/test_agent_context.py b/tests/control_plane/test_agent_context.py index 0c7db75c0..c14a29b00 100644 --- a/tests/control_plane/test_agent_context.py +++ b/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 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", + } diff --git a/tests/control_plane_ts/agent_context.test.ts b/tests/control_plane_ts/agent_context.test.ts index 772deda10..db8ac5437 100644 --- a/tests/control_plane_ts/agent_context.test.ts +++ b/tests/control_plane_ts/agent_context.test.ts @@ -188,9 +188,60 @@ 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")); + + 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); +});