From fa11e47524470f939e98fac19b067ae00bfda13a Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Mon, 19 Oct 2026 04:14:02 -0700 Subject: [PATCH 1/3] Allow explicitly configured controllers to use approved tool dispatch --- amplifier_module_loop_live/dispatch.py | 15 +++++- amplifier_module_loop_live/orchestrator.py | 3 ++ docs/programmatic-callers.md | 19 +++++++ tests/test_dispatch_callers.py | 59 ++++++++++++++++++++++ 4 files changed, 94 insertions(+), 2 deletions(-) create mode 100644 docs/programmatic-callers.md create mode 100644 tests/test_dispatch_callers.py diff --git a/amplifier_module_loop_live/dispatch.py b/amplifier_module_loop_live/dispatch.py index 1acf87e..b92f33b 100644 --- a/amplifier_module_loop_live/dispatch.py +++ b/amplifier_module_loop_live/dispatch.py @@ -112,7 +112,9 @@ def __init__(self, dispatcher, scope): name for name in scope.tools if name not in FORBIDDEN + and name not in dispatcher.callers and getattr(scope.tools[name], "name", name) not in FORBIDDEN + and getattr(scope.tools[name], "name", name) not in dispatcher.callers and (allowlist is None or name in allowlist) ) self.context = { @@ -130,6 +132,7 @@ def _valid(self, expected_scope=None): if ( self.closed or not self.dispatcher.enabled + or self.scope.call.name not in self.dispatcher.callers or not self.scope.active or not self.scope.run.active or run is not self.scope.run @@ -259,6 +262,14 @@ def __init__(self, owner): def enabled(self): return self.owner.config.get("programmatic_dispatch", False) is True + @property + def callers(self): + """Host-selected controllers; ordinary configuration permits only tool_exec.""" + names = self.owner.config.get("programmatic_callers", ["tool_exec"]) + if not isinstance(names, list) or any(not isinstance(name, str) for name in names): + return frozenset() + return frozenset(name for name in names if name == "tool_exec" or name not in FORBIDDEN) + @contextmanager def run_scope(self, coordinator): run = Run(self, coordinator, str(uuid4())) @@ -290,13 +301,13 @@ def bind(self): if ( not self.enabled or scope is None - or scope.call.name != "tool_exec" + or scope.call.name not in self.callers or not scope.active or not scope.run.active or scope.run.owner is not self or scope.run.coordinator is None ): - raise RuntimeError("No active authorized tool_exec call") + raise RuntimeError("No active authorized programmatic caller") lease = DispatchLease(self, scope) lease._valid() return lease diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py index 9a5bc17..add6763 100644 --- a/amplifier_module_loop_live/orchestrator.py +++ b/amplifier_module_loop_live/orchestrator.py @@ -535,6 +535,9 @@ async def mount(coordinator, config=None): coordinator.register_capability("live.continuation_guard_supported", True) coordinator.register_capability("tools.dispatch", loop.approved_dispatch) coordinator.register_capability("conversation.provider_pin", ConversationProviderPin(loop, coordinator)) + # Portable consumers reuse the current ordinary provider (including a host + # pin) without importing this loop or reading its private selection state. + coordinator.register_capability("model.current", lambda: loop._select_provider(coordinator.get("providers"))) async def failed(event, data): # Older Core versions omit the category. Never forward their raw error. reason = data.get("reason_code") diff --git a/docs/programmatic-callers.md b/docs/programmatic-callers.md new file mode 100644 index 0000000..15c4247 --- /dev/null +++ b/docs/programmatic-callers.md @@ -0,0 +1,19 @@ +# Explicit controller callers + +`tools.dispatch` continues to run every inner call through the ordinary streaming +engine, its pre/post hooks, approvals, cancellation and execution locks. The host +may set `programmatic_callers: [tool_exec, ui_use]` alongside +`programmatic_dispatch: true` to permit a mounted bounded controller to bind an +active-call lease. The default remains `[tool_exec]`; ambient capabilities do not +authorize another caller. Reusable behaviors do not select an orchestrator. + +The caller must have an explicitly permitted active call identity. Callers, delegate, task and +live_job cannot be nested targets. Changing the caller policy revokes outstanding +leases. `programmatic_allowed_tools` still restricts targets. No raw execution +fallback is allowed when dispatch is absent. Consumers distinguish denial, failed +execution and unknown outcomes and do not retry an unconfirmed effect. + +`model.current()` returns the currently selected ordinary provider using this +loop's normal selection and host pin. Portable consumers can use its public +`complete` protocol for bounded selection or text writing. The capability does +not select a decision backend, change routing policy, or start a model request. diff --git a/tests/test_dispatch_callers.py b/tests/test_dispatch_callers.py new file mode 100644 index 0000000..e8b22d8 --- /dev/null +++ b/tests/test_dispatch_callers.py @@ -0,0 +1,59 @@ +"""New controllers use exactly the ordinary engine/approval lease, opt-in only.""" +import json +import pytest +from amplifier_core import HookResult +from amplifier_core.message_models import ToolCall +from test_approved_dispatch import environment, Client + + +async def invoke(environment, name="ui_use", handler=None): + _, coordinator, loop, _ = environment + client = Client(coordinator, handler) + client.name = name + await coordinator.mount("tools", client, name=name) + with loop.approved_dispatch.run_scope(coordinator): + _, _, result = await loop._execute_tool_only(ToolCall(id="ui-parent", name=name, arguments={}), + coordinator.get("tools"), coordinator.hooks, "ui-group", coordinator) + return result, client + + +async def test_nondefault_caller_cannot_bind_without_host_opt_in(environment): + result, client = await invoke(environment) + assert client.lease is None + assert environment[3].calls == 0 + assert "authorized" in result + + +async def test_selected_caller_preserves_hook_denial_and_parent_identity(environment): + _, coordinator, loop, echo = environment + loop.config["programmatic_callers"] = ["tool_exec", "ui_use"] + observed = [] + async def deny(event, data): + if data["tool_name"] == "read": + observed.append(data) + return HookResult(action="deny", reason="host denied") + return HookResult() + coordinator.hooks.register("tool:pre", deny) + result, client = await invoke(environment) + assert json.loads(result)["output"]["status"] == "denied" + assert echo.calls == 0 and observed[0]["parent_tool_call_id"] == "ui-parent" + assert "ui_use" not in client.lease.tools and "tool_exec" not in client.lease.tools + + +async def test_caller_policy_change_revokes_lease(environment): + _, _, loop, echo = environment + loop.config["programmatic_callers"] = ["ui_use"] + async def revoke(lease): + loop.config["programmatic_callers"] = [] + with pytest.raises(RuntimeError): await lease.call("read", {}) + return {"revoked": True} + await invoke(environment, handler=revoke) + assert echo.calls == 0 + + +async def test_selected_caller_runs_normal_tool_and_cannot_recurse(environment): + _, _, loop, echo = environment + loop.config["programmatic_callers"] = ["ui_use"] + result, client = await invoke(environment) + assert json.loads(result)["output"]["status"] == "completed" and echo.calls == 1 + assert "ui_use" not in client.lease.tools From a1d98d41b12efb9eb595b8493f39e7cdb47d2a4c Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Mon, 19 Oct 2026 08:39:03 -0700 Subject: [PATCH 2/3] Reuse streaming programmatic execution and inference capabilities --- amplifier_module_loop_live/dispatch.py | 357 ++------------------- amplifier_module_loop_live/orchestrator.py | 18 +- docs/programmatic-callers.md | 17 +- 3 files changed, 43 insertions(+), 349 deletions(-) diff --git a/amplifier_module_loop_live/dispatch.py b/amplifier_module_loop_live/dispatch.py index b92f33b..922b854 100644 --- a/amplifier_module_loop_live/dispatch.py +++ b/amplifier_module_loop_live/dispatch.py @@ -1,334 +1,23 @@ -"""Active-call dispatch leases for portable programmatic orchestration. - -The ordinary streaming engine remains the execution and hook authority. This -adapter adds scope, nested attribution, and structured outcomes, never a second -unchecked tool execution path. -""" - -import asyncio -import json -from contextlib import contextmanager -from contextvars import ContextVar -from dataclasses import dataclass -from uuid import uuid4 - -from amplifier_core.message_models import ToolCall -from amplifier_module_loop_streaming import StreamingOrchestrator - -_RUN = ContextVar("approved_dispatch_run", default=None) -_CALL = ContextVar("approved_dispatch_call", default=None) -FORBIDDEN = frozenset({"tool_exec", "delegate", "task", "live_job"}) - - -@dataclass -class Run: - owner: object - coordinator: object - identity: str - active: bool = True - - -@dataclass -class CallScope: - run: Run - call: object - tools: dict - hooks: object - parallel_group_id: str | None - active: bool = True - - -class GuardedTool: - """Serialize actual execution, after the engine has processed approvals.""" - - def __init__(self, tool, lock, observation=None, authority=None): - self.tool, self.lock, self.observation = tool, lock, observation - self.authority = authority - - def __getattr__(self, name): - return getattr(self.tool, name) - - async def execute(self, arguments): - async with self.lock: - if self.authority: - self.authority() - if self.observation is not None: - self.observation["started"] = True - result = await self.tool.execute(arguments) - if self.observation is not None: - self.observation["raw_success"] = getattr(result, "success", None) - return result - - -class AttributedHooks: - def __init__(self, hooks, source, observation): - self.hooks, self.source, self.observation = hooks, source, observation - - async def emit(self, event, data): - enriched = { - **data, - "parent_tool_call_id": self.source["parent_tool_call_id"], - "dispatch_source": dict(self.source), - } - result = await self.hooks.emit(event, enriched) - if event == "tool:post": - self.observation["post_seen"] = True - returned = result.data.get("result") if result and result.data else None - self.observation["post_result"] = ( - returned if returned is not None else data.get("result") - ) - elif event == "tool:error": - self.observation["engine_error"] = True - return result - - def __getattr__(self, name): - return getattr(self.hooks, name) - - -class AttributedCoordinator: - def __init__(self, coordinator, observation): - object.__setattr__(self, "coordinator", coordinator) - object.__setattr__(self, "observation", observation) - - def __getattr__(self, name): - return getattr(self.coordinator, name) - - def __setattr__(self, name, value): - setattr(self.coordinator, name, value) - - async def process_hook_result(self, result, event, name): - result = await self.coordinator.process_hook_result(result, event, name) - if result.action == "deny": - self.observation["denied"] = True - return result - - -class DispatchLease: - def __init__(self, dispatcher, scope): - self.dispatcher, self.scope = dispatcher, scope - self.closed = False - allowlist = dispatcher.owner.config.get("programmatic_allowed_tools") - self.tools = tuple( - name - for name in scope.tools - if name not in FORBIDDEN - and name not in dispatcher.callers - and getattr(scope.tools[name], "name", name) not in FORBIDDEN - and getattr(scope.tools[name], "name", name) not in dispatcher.callers - and (allowlist is None or name in allowlist) - ) - self.context = { - "session_id": str(scope.run.coordinator.session_id), - "run_id": scope.run.identity, - "parent_tool_call_id": scope.call.id, - } - - def close(self): - self.closed = True - - def _valid(self, expected_scope=None): - run = _RUN.get() - scope = _CALL.get() - if ( - self.closed - or not self.dispatcher.enabled - or self.scope.call.name not in self.dispatcher.callers - or not self.scope.active - or not self.scope.run.active - or run is not self.scope.run - or scope is not (expected_scope or self.scope) - or run.owner is not self.dispatcher - or run.coordinator is not self.scope.run.coordinator - ): - raise RuntimeError( - "Dispatch lease is not active in this session/run/parent call" - ) - if run.coordinator.cancellation.is_immediate: - raise RuntimeError("Immediate host cancellation is active") - activation = run.coordinator.get_capability("live.activation") - if activation is not None and not activation.current_valid: - raise RuntimeError("Host activation is no longer valid") - - async def wait_cancelled(self): - """Observe the real host token even across an async foreign-runtime bridge.""" - while True: - if self.scope.run.coordinator.cancellation.is_immediate: - return "host_cancel_requested" - try: - self._valid() - except RuntimeError: - return "lease_revoked" - await asyncio.sleep(0.05) - - async def call(self, name, arguments, *, request_id=None): - self._valid() - if name not in self.tools or name in FORBIDDEN: - raise ValueError("Tool is not available through this lease") - if not isinstance(arguments, dict) or arguments.get("async") is True: - raise ValueError("Background tool dispatch is unsupported") - call = ToolCall(id=str(uuid4()), name=name, arguments=arguments) - source = { - **self.context, - "tool_call_id": call.id, - "tool_name": name, - "request_id": request_id, - } - observation = {} - tools = self.dispatcher.wrap_tools( - self.scope.tools, observed_name=name, observation=observation - ) - hooks = AttributedHooks(self.scope.hooks, source, observation) - coordinator = AttributedCoordinator(self.scope.run.coordinator, observation) - # Always take the existing ordinary engine path. In particular, never - # call a mounted tool directly when hooks or authority are unavailable. - try: - with self.dispatcher.call_scope( - call, tools, hooks, self.scope.parallel_group_id - ) as child_scope: - tools = self.dispatcher.wrap_tools( - tools, - observed_name=name, - observation=observation, - authority=lambda: self._valid(child_scope), - ) - _, _, content = await StreamingOrchestrator._execute_tool_only( - self.dispatcher.owner, - call, - tools, - hooks, - self.scope.parallel_group_id or self.scope.call.id, - coordinator, - ) - except asyncio.CancelledError: - await self.scope.hooks.emit( - "tool:dispatch_cancelled", - { - "tool_name": name, - "tool_call_id": call.id, - "parent_tool_call_id": self.scope.call.id, - "dispatch_source": source, - "outcome": "unknown", - "effects": "not_rolled_back", - }, - ) - raise - self._valid() - if observation.get("denied"): - status, success, output = ( - "denied", - False, - "Tool execution denied by host policy", - ) - else: - post = observation.get("post_result") - confirmed = post.get("success") if isinstance(post, dict) else None - if observation.get("engine_error"): - status, success = "failed", False - elif observation.get("post_seen") and confirmed is True: - status, success = "completed", True - elif observation.get("post_seen") and confirmed is False: - status, success = "failed", False - else: - status, success = "unknown", None - try: - output = json.loads(content) - # The core hook transport may copy result dictionaries, making - # the ordinary engine serialize the full post-hook ToolResult. - # Unwrap only when it is exactly the captured confirmed envelope. - if ( - isinstance(post, dict) - and isinstance(post.get("success"), bool) - and output == post - ): - output = post.get("output") - except (ValueError, TypeError): - output = content - return { - "status": status, - "success": success, - "output": output, - "source": source, - } - - -class ApprovedDispatch: - version = 1 - - def __init__(self, owner): - self.owner = owner - self.locks = {} - - @property - def enabled(self): - return self.owner.config.get("programmatic_dispatch", False) is True - - @property - def callers(self): - """Host-selected controllers; ordinary configuration permits only tool_exec.""" - names = self.owner.config.get("programmatic_callers", ["tool_exec"]) - if not isinstance(names, list) or any(not isinstance(name, str) for name in names): - return frozenset() - return frozenset(name for name in names if name == "tool_exec" or name not in FORBIDDEN) - - @contextmanager - def run_scope(self, coordinator): - run = Run(self, coordinator, str(uuid4())) - token = _RUN.set(run) - try: - yield - finally: - run.active = False - _RUN.reset(token) - - @contextmanager - def call_scope(self, call, tools, hooks, parallel_group_id): - run = _RUN.get() - scope = ( - CallScope(run, call, tools, hooks, parallel_group_id) - if run and run.owner is self - else None - ) - token = _CALL.set(scope) - try: - yield scope - finally: - if scope: - scope.active = False - _CALL.reset(token) - - def bind(self): - scope = _CALL.get() - if ( - not self.enabled - or scope is None - or scope.call.name not in self.callers - or not scope.active - or not scope.run.active - or scope.run.owner is not self - or scope.run.coordinator is None - ): - raise RuntimeError("No active authorized programmatic caller") - lease = DispatchLease(self, scope) - lease._valid() - return lease - - def wrap_tools(self, tools, observed_name=None, observation=None, authority=None): - wrapped = {} - for name, tool in tools.items(): - raw = tool.tool if isinstance(tool, GuardedTool) else tool - allowlist = self.owner.config.get("programmatic_allowed_tools") - if name != "tool_exec" and ( - name in FORBIDDEN - or getattr(raw, "name", name) in FORBIDDEN - or (allowlist is not None and name not in allowlist) - ): - wrapped[name] = raw - continue - lock = self.locks.setdefault(id(raw), asyncio.Lock()) - wrapped[name] = GuardedTool( - raw, - lock, - observation if name == observed_name else None, - authority if name == observed_name else None, - ) - return wrapped +"""Compatibility imports; execution authority is shared with loop-streaming.""" + +from amplifier_module_loop_streaming.dispatch import ( + FORBIDDEN, + ApprovedDispatch, + AttributedCoordinator, + AttributedHooks, + CallScope, + DispatchLease, + GuardedTool, + Run, +) + +__all__ = [ + "FORBIDDEN", + "ApprovedDispatch", + "AttributedCoordinator", + "AttributedHooks", + "CallScope", + "DispatchLease", + "GuardedTool", + "Run", +] diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py index add6763..4971d48 100644 --- a/amplifier_module_loop_live/orchestrator.py +++ b/amplifier_module_loop_live/orchestrator.py @@ -19,7 +19,6 @@ from .job_store import serialize_result from .host import HostAdapter, AttachmentContext from .scope import HOST_ADAPTER -from .dispatch import ApprovedDispatch from .failures import ManagerTurnError, turn_failure _WAKE = "[loop-live inbox wake]" @@ -35,7 +34,12 @@ def __init__(self, config=None): self.load_failures = [] self.running = False self.root_provider = None - self.approved_dispatch = ApprovedDispatch(self) + # Live activation belongs to this adapter. The shared execution service + # accepts only a generic host-validity callback. + def dispatch_context_valid(coordinator): + activation = coordinator.get_capability("live.activation") + return activation is None or activation.current_valid is True + self.approved_dispatch.validate_context = dispatch_context_valid async def _execute_guarded_goal(self, prompt, context, providers, tools, hooks, coordinator): # An explicit input always receives its first turn. Only automatic goal @@ -442,13 +446,6 @@ def cancel_job(job): job["task"].cancel() async def _execute_tool_only(self, tool_call, tools, hooks, parallel_group_id, coordinator=None): - if not self.approved_dispatch.enabled: - return await self._execute_tool_scoped(tool_call, tools, hooks, parallel_group_id, coordinator) - tools = self.approved_dispatch.wrap_tools(tools) - with self.approved_dispatch.call_scope(tool_call, tools, hooks, parallel_group_id): - return await self._execute_tool_scoped(tool_call, tools, hooks, parallel_group_id, coordinator) - - async def _execute_tool_scoped(self, tool_call, tools, hooks, parallel_group_id, coordinator=None): arguments = tool_call.arguments or {} eligible = tool_call.name in self.config.get("background_tools", ["delegate"]) default = bool(self.config.get("background_delegate") and tool_call.name == "delegate") @@ -534,6 +531,9 @@ async def mount(coordinator, config=None): coordinator.register_capability("session.steer", loop.steer) coordinator.register_capability("live.continuation_guard_supported", True) coordinator.register_capability("tools.dispatch", loop.approved_dispatch) + coordinator.register_capability("model.complete", loop.approved_dispatch.complete) + coordinator.register_contributor("observability.events", "loop-live-programmatic", + lambda: ["model:complete", "tool:dispatch_cancelled"]) coordinator.register_capability("conversation.provider_pin", ConversationProviderPin(loop, coordinator)) # Portable consumers reuse the current ordinary provider (including a host # pin) without importing this loop or reading its private selection state. diff --git a/docs/programmatic-callers.md b/docs/programmatic-callers.md index 15c4247..2020930 100644 --- a/docs/programmatic-callers.md +++ b/docs/programmatic-callers.md @@ -4,8 +4,11 @@ engine, its pre/post hooks, approvals, cancellation and execution locks. The host may set `programmatic_callers: [tool_exec, ui_use]` alongside `programmatic_dispatch: true` to permit a mounted bounded controller to bind an -active-call lease. The default remains `[tool_exec]`; ambient capabilities do not -authorize another caller. Reusable behaviors do not select an orchestrator. +active-call lease. Without explicit caller configuration, a portable module may +declare its caller and bounded target list in `tools.dispatch.registrations` at +mount. Explicit host disable/caller/target restrictions take precedence. This +trusted module declaration is separate from model output or ambient credentials. +Reusable behaviors do not select an orchestrator. The caller must have an explicitly permitted active call identity. Callers, delegate, task and live_job cannot be nested targets. Changing the caller policy revokes outstanding @@ -13,7 +16,9 @@ leases. `programmatic_allowed_tools` still restricts targets. No raw execution fallback is allowed when dispatch is absent. Consumers distinguish denial, failed execution and unknown outcomes and do not retry an unconfirmed effect. -`model.current()` returns the currently selected ordinary provider using this -loop's normal selection and host pin. Portable consumers can use its public -`complete` protocol for bounded selection or text writing. The capability does -not select a decision backend, change routing policy, or start a model request. +The implementation now lives in loop-streaming; this module reuses it rather than +maintaining a second execution engine. `model.complete(ChatRequest)` provides +bounded, attributed ordinary inference with provider request policy, input-budget +checks, call counting and cancellation. The legacy `model.current()` accessor is +retained for compatibility; UI-use no longer calls a provider through that raw +accessor. The service selects no decision backend and changes no routing policy. From d38157796843c7aea65d5afa74084429d5f3f23f Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Thu, 22 Oct 2026 01:03:25 -0700 Subject: [PATCH 3/3] Resolve accepted Streaming dispatcher in lockfile --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index 6316a38..3db1384 100644 --- a/uv.lock +++ b/uv.lock @@ -61,7 +61,7 @@ dev = [ [[package]] name = "amplifier-module-loop-streaming" version = "1.0.0" -source = { git = "https://github.com/microsoft/amplifier-module-loop-streaming?rev=main#9e5d4ff07c7f57fbdf233b869d94e18c910d83b9" } +source = { git = "https://github.com/microsoft/amplifier-module-loop-streaming?rev=main#e04bba7e85bff9836bfc48cd8ac1db676f4f5fab" } [[package]] name = "annotated-types"