From 850025f68b950ff15cafe07bd4b7570bfa4b6a15 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:48:48 +0800 Subject: [PATCH] Limit work context planning to selection facts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../control_plane/work_items/context_readback.py | 15 ++++++++++++++- .../test_work_acceptance_context.py | 16 +++++++++++++++- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/work_items/context_readback.py b/loopx/control_plane/work_items/context_readback.py index 5a9f427273..488771da59 100644 --- a/loopx/control_plane/work_items/context_readback.py +++ b/loopx/control_plane/work_items/context_readback.py @@ -104,7 +104,20 @@ def attach_work_context(payload: dict[str, Any], *, registry_path: Path, agent_id = (payload.get("agent_identity") or {}).get("agent_id") if not goal_id: return - plan = effect_runtime_result("work_item.context.plan", {"packet": payload}) + # The typed planner needs the selection inputs, not the admission body + # that the public reader and final capture continue to own. + cli = interaction.get("cli_channel") + orchestration = payload.get("task_orchestration_contract") + plan = effect_runtime_result("work_item.context.plan", {"packet": { + "selected_todo": packet_selected, + "effective_action": payload.get("effective_action"), + "task_orchestration_contract": { + key: orchestration.get(key) for key in ("schema_version", "mode", "primary_todo_id") + } if isinstance(orchestration, dict) else {}, + "interaction_contract": {"cli_channel": { + "selection_required": cli.get("selection_required"), + } if isinstance(cli, dict) else {}}, + }}) selected = plan.get("selected_todo") if isinstance(selected, dict) and isinstance(selected_text_snapshot, str): selected["_context_text_sha256"] = selected_text_snapshot diff --git a/tests/control_plane/test_work_acceptance_context.py b/tests/control_plane/test_work_acceptance_context.py index fcb915cce8..d61b3a71b0 100644 --- a/tests/control_plane/test_work_acceptance_context.py +++ b/tests/control_plane/test_work_acceptance_context.py @@ -166,16 +166,30 @@ def cli(*arguments): def unavailable(**_kwargs): raise OSError("acceptance source unavailable") + planner_sources = [] + runtime_result = context_readback.effect_runtime_result + + def observe_planner_source(method, params, **options): + if method == "work_item.context.plan": + planner_sources.append(params["packet"]) + return runtime_result(method, params, **options) + with monkeypatch.context() as fault: fault.setattr(context_readback, "inspect_goal_acceptance", unavailable) + fault.setattr(context_readback, "effect_runtime_result", observe_planner_source) output = io.StringIO() with contextlib.redirect_stdout(output): code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", - "quota", "should-run", "--goal-id", goal, "--agent-id", "agent-a", ]) + "quota", "should-run", "--goal-id", goal, "--agent-id", "agent-a", "--context-delivery", "off"]) failed = json.loads(output.getvalue())["interaction_contract"]["agent_channel"] assert code == 0 and failed["delivery_allowed"] is False assert failed["work_context"]["complete"] is False assert any(row.get("source") == "goal_acceptance" for row in failed["required_reads"]) + assert planner_sources + for source in planner_sources: + assert set(source) == {"selected_todo", "task_orchestration_contract", "effective_action", "interaction_contract"} + assert set(source["interaction_contract"]["cli_channel"]) <= {"selection_required"} + assert set(source["task_orchestration_contract"]) <= {"schema_version", "mode", "primary_todo_id"} recovered = cli("quota", "should-run", "--goal-id", goal, "--agent-id", "agent-a", ) assert recovered["interaction_contract"]["agent_channel"]["work_context"]["complete"] is True