"""Platform-neutral tool host using the existing transactional adapter. The host injects verified login and scheduler lookups; neither is a model tool. Business Agent labels are dispatch metadata only, never authorization identities or budget keys. Old module/class names, encrypted rows and request IDs remain compatible. No second payment implementation, credential discovery or Agent runtime is created here. """ from __future__ import annotations from dataclasses import dataclass import re # Keep the proven implementation and old import path compatible. These names # are the generic public port; the historical table/request prefixes are not # changed because doing so could abandon journals and duplicate purchase intent. from alvin_adapter import ( AdapterError, AlvinAgentTools as _ExistingAgentTools, EncryptedStore, SubjectContextProvider, TaskBinding, TrustedContext, TOOLS, partner_subject_digest, ) class AgentTools(_ExistingAgentTools): """Existing tool semantics plus explicit original-preview startup control.""" async def call(self, context, name, arguments): result = await super().call(context, name, arguments) if name == "tools/list": for item in result.get("tools", []): if item.get("name") == "execute_clink_purchase": # Recovery is a trusted host/scheduler decision after an # authoritative original-state read, not a model argument. schema = item.get("inputSchema", {}) schema.get("properties", {}).pop("startup_action", None) if "required" in schema: schema["required"] = [key for key in schema["required"] if key != "startup_action"] return result async def recover_startup(self, context, action): if action not in {"resume", "terminate"}: raise AdapterError("STARTUP_RECOVERY_NOT_AVAILABLE") self._context(context, active=action == "resume") owner = self._identity(context) record = self._intent(context) if not record or not record.get("execute_claimed") or not record.get("preview_id"): raise AdapterError("STARTUP_RECOVERY_NOT_AVAILABLE") original = await self._invoke(owner, "get_clink_purchase_preview", {"preview_id": record["preview_id"]}) data = original.get("data", {}) if original.get("status") == "ok" else {} startup = data.get("startup", {}) if data.get("preview_id") != record["preview_id"] or startup.get("protocol") != "original_preview_startup_v1": raise AdapterError("STARTUP_RECOVERY_NOT_AVAILABLE") if startup.get("state") in {"admitted", "terminated"}: self._record_result(context, original) return await self._recover(context, owner) if startup["state"] == "admitted" else original if startup.get("state") != "prepared" or action + "_original" not in startup.get("next_actions", []): raise AdapterError("STARTUP_RECOVERY_NOT_AVAILABLE") await self._access_token(owner) def claim(row): row["intents"][context.intent_id]["startup_action_requested"] = action return row self._task_change(context, claim, active=action == "resume") # Core/Marketplace atomically admit or terminate this exact preview. # A lost response retains the journal and requires another explicit # original-state check, never a timer-based resend or replacement ID. try: result = await self._invoke(owner, "execute_clink_purchase", { "preview_id": record["preview_id"], "startup_action": action}) self._record_result(context, result) return result except AdapterError: return self._unknown(context) async def resume_startup(self, context): return await self.recover_startup(context, "resume") async def terminate_startup(self, context): return await self.recover_startup(context, "terminate") @dataclass(frozen=True, repr=False) class DispatchBinding: subject_id: str business_agent_id: str task_id: str intent_id: str generation: int class MultiUserHost: """Executable integration port, not an anonymous public multi-user API. login_subject(request) must use the host's already verified login; it must not simply read a user_id from an HTTP body. dispatch_lookup(handle) returns a server-stored task assignment. task_owner remains the authority for task cancellation and generation. A shared Agent role cannot select a different user's connection. Each host uses its own authentication framework. """ def __init__(self, client, store, *, login_subject, dispatch_lookup, task_owner, environment): if not callable(login_subject) or not callable(dispatch_lookup): raise AdapterError("TRUSTED_HOST_CONFIGURATION_REQUIRED") self.client, self.store = client, store self.login_subject, self.dispatch_lookup = login_subject, dispatch_lookup self.tools = AgentTools(client, store, task_owner=task_owner, environment=environment) def context(self, request, dispatch_handle): if type(dispatch_handle) is not str or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,200}", dispatch_handle): raise AdapterError("TRUSTED_DISPATCH_REQUIRED") try: subject = self.login_subject(request) binding = self.dispatch_lookup(dispatch_handle) except Exception: raise AdapterError("TRUSTED_DISPATCH_REQUIRED") from None if (type(subject) is not str or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,200}", subject) or type(binding) is not DispatchBinding or binding.subject_id != subject or type(binding.business_agent_id) is not str or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,200}", binding.business_agent_id)): raise AdapterError("TRUSTED_DISPATCH_REQUIRED") context = TrustedContext(subject, binding.task_id, binding.intent_id, binding.generation) self.tools._context(context) return context def connection_owner(self, request, dispatch_handle): return self.tools.ensure_connection(self.context(request, dispatch_handle)) async def connect(self, request, dispatch_handle): context = self.context(request, dispatch_handle) self.tools._task_change(context, lambda row: row, active=True) owner = self.tools.ensure_connection(context) return await self.client.begin(owner) async def callback(self, request, dispatch_handle, callback_parameters): context = self.context(request, dispatch_handle) self.tools._task_change(context, lambda row: row, active=True) owner = self.tools.connection_owner(context) await self.client.callback(owner, callback_parameters) # Late completion cannot restart the task; a subsequent dispatch always # rechecks the persistent cancellation/generation fence. self.tools._task_change(context, lambda row: row, active=True) return {"status": "connected", "automatic_dispatch": False} async def call(self, request, dispatch_handle, name, arguments): return await self.tools.call(self.context(request, dispatch_handle), name, arguments) def cancel(self, request, dispatch_handle): """Cancel the shared task, not an Agent or the user's common Mandate.""" return self.tools.cancel(self.context(request, dispatch_handle)) async def result(self, request, dispatch_handle): return await self.tools.purchase_result(self.context(request, dispatch_handle)) async def resume_startup(self, request, dispatch_handle): return await self.tools.resume_startup(self.context(request, dispatch_handle)) async def terminate_startup(self, request, dispatch_handle): return await self.tools.terminate_startup(self.context(request, dispatch_handle))