diff --git a/README.md b/README.md index 7929021..c6c48ea 100644 --- a/README.md +++ b/README.md @@ -83,7 +83,7 @@ await client.close() A protocol plugin registers itself in one of two registries, and the choice decides who its state belongs to. -An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that is *meant* to be shared — a credential cache, a pooled HTTP session, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it. +An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that is *meant* to be shared — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it. A protocol whose state **must not be shared between clients** registers a **factory** instead: live sessions or connections keyed per manual, child processes — anything one client's use or `close()` would take away from another. Each `UtcpClient` calls the factory once — at creation, or on first use if it was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`: @@ -95,7 +95,7 @@ register_communication_protocol_factory("custom_type", CustomCommunicationProtoc This is what makes "a client per tenant / per user / per pooled connection" actually isolate them, rather than every client reaching into one shared instance. A type registered as a factory wins over the same type registered as an instance, so a plugin migrates by moving its registration from one call to the other and callers change nothing. -`utcp-mcp` and `utcp-websocket` register this way — MCP sessions (and, for stdio, child processes) and per-manual WebSocket connections belong to the client that opened them. `utcp-http` (HTTP, SSE, streamable HTTP) and `utcp-gql` stay shared instances: what they keep is an OAuth token cache and a pooled HTTP session, both meant to be reused across clients. +`utcp-mcp` and `utcp-websocket` register this way — MCP sessions (and, for stdio, child processes) and per-manual WebSocket connections belong to the client that opened them. `utcp-http` (HTTP, SSE, streamable HTTP) and `utcp-gql` stay shared instances: the only state they keep is an OAuth token cache, which is meant to be reused across clients; every request opens its own session. `client.close()` closes the instances the client created — every one of them, even if one fails, after which the failures are raised together as `UtcpProtocolCloseError`. If `UtcpClient.create` fails after creating them, they are closed the same way before the error is re-raised. diff --git a/core/README.md b/core/README.md index 7929021..c6c48ea 100644 --- a/core/README.md +++ b/core/README.md @@ -83,7 +83,7 @@ await client.close() A protocol plugin registers itself in one of two registries, and the choice decides who its state belongs to. -An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that is *meant* to be shared — a credential cache, a pooled HTTP session, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it. +An instance registered with `register_communication_protocol` is **shared by every `UtcpClient` in the process**, and so is any state it keeps. That is right for state that is *meant* to be shared — a credential cache, a registry a decorator writes into — and wrong for state that belongs to one client. A shared instance lives as long as the process; no client closes it. A protocol whose state **must not be shared between clients** registers a **factory** instead: live sessions or connections keyed per manual, child processes — anything one client's use or `close()` would take away from another. Each `UtcpClient` calls the factory once — at creation, or on first use if it was registered later — so each client gets its own instance, its own connections, and its own teardown on `close()`: @@ -95,7 +95,7 @@ register_communication_protocol_factory("custom_type", CustomCommunicationProtoc This is what makes "a client per tenant / per user / per pooled connection" actually isolate them, rather than every client reaching into one shared instance. A type registered as a factory wins over the same type registered as an instance, so a plugin migrates by moving its registration from one call to the other and callers change nothing. -`utcp-mcp` and `utcp-websocket` register this way — MCP sessions (and, for stdio, child processes) and per-manual WebSocket connections belong to the client that opened them. `utcp-http` (HTTP, SSE, streamable HTTP) and `utcp-gql` stay shared instances: what they keep is an OAuth token cache and a pooled HTTP session, both meant to be reused across clients. +`utcp-mcp` and `utcp-websocket` register this way — MCP sessions (and, for stdio, child processes) and per-manual WebSocket connections belong to the client that opened them. `utcp-http` (HTTP, SSE, streamable HTTP) and `utcp-gql` stay shared instances: the only state they keep is an OAuth token cache, which is meant to be reused across clients; every request opens its own session. `client.close()` closes the instances the client created — every one of them, even if one fails, after which the failures are raised together as `UtcpProtocolCloseError`. If `UtcpClient.create` fails after creating them, they are closed the same way before the error is re-raised. diff --git a/core/src/utcp/implementations/utcp_client_implementation.py b/core/src/utcp/implementations/utcp_client_implementation.py index c227b63..e494b6e 100644 --- a/core/src/utcp/implementations/utcp_client_implementation.py +++ b/core/src/utcp/implementations/utcp_client_implementation.py @@ -25,6 +25,12 @@ logger = logging.getLogger(__name__) + +def _sanitize_manual_name(name: str) -> str: + """The name a manual is registered under: every non-word character becomes an underscore.""" + return re.sub(r'[^\w]', '_', name) + + class UtcpClientImplementation(UtcpClient): """REQUIRED Implementation of the `UtcpClient` interface. @@ -151,9 +157,14 @@ async def create( client = cls(config, DefaultVariableSubstitutor(), root_dir) # Everything from here on can fail, and the caller never receives a - # client it could close — so whatever this client CREATED is closed - # here before the failure is re-raised. Only what it created: the shared - # instances are in use by every other client and are not this one's. + # client it could close — so whatever this client CREATED is undone + # here before the failure is re-raised: the manuals this attempt + # registered are removed (the tool repository may be caller-supplied + # and shared), then the protocol instances it created are closed. + # Exactly what it registered — recorded as each registration + # succeeds, never inferred from repository state, which another client + # sharing the repository may have changed in the meantime. + registered_by_this_attempt: List[str] = [] try: client._adopt_factory_protocols() @@ -165,15 +176,29 @@ async def create( # Load the manuals if any if config.manual_call_templates: - await client.register_manuals(config.manual_call_templates) + results = await client._register_each(config.manual_call_templates, registered_by_this_attempt) + failures = [result for result in results if isinstance(result, BaseException)] + if failures: + raise failures[0] except BaseException: + # Rollback first (deregistration needs the protocols open), then + # close — in a finally, so a cancellation that lands during the + # rollback still closes what this client created and then reaches + # the caller, as a cancellation must. try: - await client._close_owned_protocols() - except Exception: - # The initialization failure stays the error the caller sees; - # the cleanup failure is reported rather than swallowed, because - # the caller has no client through which to retry it. - logger.error("UtcpClient.create failed, and closing the protocols it had created failed too", exc_info=True) + for name in registered_by_this_attempt: + try: + await client.deregister_manual(name) + except Exception: + logger.error(f"UtcpClient.create failed, and removing the manual '{name}' it had registered failed too", exc_info=True) + finally: + try: + await client._close_owned_protocols() + except Exception: + # The initialization failure stays the error the caller sees; + # the cleanup failure is reported rather than swallowed, because + # the caller has no client through which to retry it. + logger.error("UtcpClient.create failed, and closing the protocols it had created failed too", exc_info=True) raise return client @@ -204,7 +229,7 @@ async def register_manual(self, manual_call_template: CallTemplate) -> RegisterM ValueError: If manual name is already registered or communication protocol is not found. """ # Replace all non-word characters with underscore - manual_call_template.name = re.sub(r'[^\w]', '_', manual_call_template.name) + manual_call_template.name = _sanitize_manual_name(manual_call_template.name) if await self.config.tool_repository.get_manual(manual_call_template.name) is not None: raise ValueError(f"Manual {manual_call_template.name} already registered, please use a different name or deregister the existing manual") manual_call_template = self._substitute_call_template_variables(manual_call_template, manual_call_template.name) @@ -247,41 +272,55 @@ async def register_manuals(self, manual_call_templates: List[CallTemplate]) -> L Returns: A list of `RegisterManualResult` instances representing the results of the registration. """ - # Create tasks for parallel CallTemplate registration - tasks = [] - for manual_call_template in manual_call_templates: - async def try_register_manual(manual_call_template=manual_call_template): - try: - result = await self.register_manual(manual_call_template) - if result.success: - logger.info(f"Successfully registered manual '{manual_call_template.name}' with {len(result.manual.tools)} tools") - else: - logger.error(f"Error registering manual '{manual_call_template.name}': {result.errors}") - return result - except UtcpVariableNotFound as e: - raise e - except Exception as e: - logger.error(f"Error registering manual '{manual_call_template.name}': {traceback.format_exc()}") - return RegisterManualResult( - manual_call_template=manual_call_template, - manual=UtcpManual(manual_version="0.0.0", tools=[]), - success=False, - errors=[traceback.format_exc()] - ) - - tasks.append(try_register_manual()) - - # Wait for EVERY registration to settle, even when one raises. The - # caller receives the failure only once no sibling registration is - # still running underneath it — create(), for one, closes the - # protocols right after, and a registration still in flight would be - # using a closed one. - results = await asyncio.gather(*tasks, return_exceptions=True) + results = await self._register_each(manual_call_templates) failures = [result for result in results if isinstance(result, BaseException)] if failures: raise failures[0] return [p for p in results if p is not None] + async def _register_each( + self, + manual_call_templates: List[CallTemplate], + registered_names: Optional[List[str]] = None, + ) -> List[Union[RegisterManualResult, BaseException]]: + """Register every template in parallel and wait for EVERY one to settle. + + A raised failure is returned in place of its result rather than + propagated at once, so no sibling registration is still running + underneath a caller that holds the failure — create(), for one, closes + the protocols right after, and a registration still in flight would be + using a closed one. + + When ``registered_names`` is given, the registered name of each manual + that succeeded is appended to it AS it succeeds, so a caller that + fails part-way knows exactly what it registered — and only that. + """ + async def try_register_manual(manual_call_template: CallTemplate): + try: + result = await self.register_manual(manual_call_template) + if result.success: + if registered_names is not None: + registered_names.append(manual_call_template.name) + logger.info(f"Successfully registered manual '{manual_call_template.name}' with {len(result.manual.tools)} tools") + else: + logger.error(f"Error registering manual '{manual_call_template.name}': {result.errors}") + return result + except UtcpVariableNotFound as e: + raise e + except Exception as e: + logger.error(f"Error registering manual '{manual_call_template.name}': {traceback.format_exc()}") + return RegisterManualResult( + manual_call_template=manual_call_template, + manual=UtcpManual(manual_version="0.0.0", tools=[]), + success=False, + errors=[traceback.format_exc()] + ) + + return await asyncio.gather( + *(try_register_manual(manual_call_template) for manual_call_template in manual_call_templates), + return_exceptions=True, + ) + async def deregister_manual(self, manual_name: str) -> bool: """REQUIRED Deregister a manual from the client. @@ -426,7 +465,7 @@ async def get_required_variables_for_manual_and_tools(self, manual_call_template Returns: A list of required variables for the manual and its tools. """ - manual_call_template.name = re.sub(r'[^\w]', '_', manual_call_template.name) + manual_call_template.name = _sanitize_manual_name(manual_call_template.name) variables_for_CallTemplate = self.variable_substitutor.find_required_variables(CallTemplateSerializer().to_dict(manual_call_template), manual_call_template.name) if len(variables_for_CallTemplate) > 0: try: diff --git a/core/src/utcp/interfaces/communication_protocol.py b/core/src/utcp/interfaces/communication_protocol.py index 56c055b..a47b229 100644 --- a/core/src/utcp/interfaces/communication_protocol.py +++ b/core/src/utcp/interfaces/communication_protocol.py @@ -31,7 +31,7 @@ class CommunicationProtocol(ABC): - `communication_protocols` holds an INSTANCE that is shared by every `UtcpClient` in the process, and so is any state it keeps. That is the right home for state that is meant to be shared (a credential cache, a - pooled HTTP session, a registry a decorator writes into). The instance + registry a decorator writes into). The instance lives as long as the process that registered it; no client closes it. - `communication_protocol_factories` holds a FACTORY, for a protocol whose state must not be shared between clients: live sessions or connections diff --git a/core/src/utcp/plugins/discovery.py b/core/src/utcp/plugins/discovery.py index d7aca50..c08d513 100644 --- a/core/src/utcp/plugins/discovery.py +++ b/core/src/utcp/plugins/discovery.py @@ -91,8 +91,8 @@ def register_communication_protocol_factory(communication_protocol_type: str, fa Use this instead of `register_communication_protocol` for a protocol whose state must not be shared between clients: live sessions or connections keyed per manual, child processes — anything one client's use or `close()` - would take away from another. (A credential cache or a pooled HTTP session - is meant to be shared and stays an instance.) Each `UtcpClient` calls the factory once — at + would take away from another. (A credential cache is meant to be shared + and stays an instance.) Each `UtcpClient` calls the factory once — at creation, or on first use if the factory is registered later — and that client's `close()` tears the instance down. A type registered as a factory wins over the same type registered as an instance. diff --git a/core/tests/client/test_client_protocol_ownership.py b/core/tests/client/test_client_protocol_ownership.py index 01f00da..b9217c8 100644 --- a/core/tests/client/test_client_protocol_ownership.py +++ b/core/tests/client/test_client_protocol_ownership.py @@ -16,7 +16,10 @@ from utcp.data.tool import JsonSchema, Tool from utcp.data.utcp_manual import UtcpManual from utcp.exceptions import UtcpProtocolCloseError, UtcpVariableNotFound +from utcp.data.utcp_client_config import UtcpClientConfig +from utcp.implementations.in_mem_tool_repository import InMemToolRepository from utcp.interfaces.communication_protocol import CommunicationProtocol +from utcp.plugins.plugin_loader import ensure_plugins_initialized from utcp.utcp_client import UtcpClient from utcp_http.http_call_template import HttpCallTemplate @@ -66,7 +69,14 @@ def http_manual(name: str) -> HttpCallTemplate: @pytest.fixture def isolated_registries(monkeypatch): - """Both registries start empty and are restored after the test.""" + """Both registries start empty and are restored after the test. + + Plugins load lazily on the first ``UtcpClient.create()`` of the session and + register into whatever dict holds the class attribute at that moment. They + are loaded here first, so their registrations land in the ORIGINAL dicts + (which the fixture restores) and never in the throwaway ones. + """ + ensure_plugins_initialized() monkeypatch.setattr(CommunicationProtocol, "communication_protocols", {}) monkeypatch.setattr(CommunicationProtocol, "communication_protocol_factories", {}) @@ -315,3 +325,124 @@ async def test_a_failed_create_closes_its_protocols_only_after_every_registratio ]}) assert events == ["registered slow_manual", "closed"] + + +class TestAFailedCreateLeavesNoManualBehind: + + @pytest.mark.asyncio + async def test_a_manual_registered_by_the_failed_attempt_is_removed_from_a_shared_repository(self, isolated_registries): + # The caller supplies (and keeps) the repository; the second manual + # registers fine before the first one's failure surfaces. create() + # raises -- and must not leave that manual behind in the caller's repo. + CommunicationProtocol.communication_protocols["http"] = RecordingProtocol() + repo = InMemToolRepository() + + with pytest.raises(UtcpVariableNotFound): + await UtcpClient.create(config=UtcpClientConfig( + tool_repository=repo, + manual_call_templates=[unresolvable_http_manual("bad_manual"), http_manual("good_manual")], + )) + + assert await repo.get_manual("good_manual") is None + assert await repo.get_tool("good_manual.ping") is None + + @pytest.mark.asyncio + async def test_a_manual_that_was_in_the_repository_before_the_attempt_survives_it(self, isolated_registries): + # "existing" is already registered by an earlier client on the same + # shared repository. This attempt names it again (that registration + # fails as a duplicate, without raising) and also fails outright on + # the bad manual. The rollback must remove only what THIS attempt + # added -- never the pre-existing manual. + CommunicationProtocol.communication_protocols["http"] = RecordingProtocol() + repo = InMemToolRepository() + earlier = await UtcpClient.create(config=UtcpClientConfig(tool_repository=repo, manual_call_templates=[http_manual("existing")])) + assert await repo.get_manual("existing") is not None + + with pytest.raises(UtcpVariableNotFound): + await UtcpClient.create(config=UtcpClientConfig( + tool_repository=repo, + manual_call_templates=[http_manual("existing"), unresolvable_http_manual("bad_manual"), http_manual("added_by_attempt")], + )) + + assert await repo.get_manual("existing") is not None + assert await repo.get_manual("added_by_attempt") is None + await earlier.close() + + +def test_plugin_registrations_survive_the_isolated_tests_above(): + # Runs after every isolated test in this file. If any of them had been the + # session's first create() and the fixture had patched the registry before + # plugins loaded, the plugin instances would have gone into the throwaway + # dict and be missing here. + ensure_plugins_initialized() + assert "http" in CommunicationProtocol.communication_protocols + assert "mcp" in CommunicationProtocol.communication_protocol_factories + + +class TestRollbackRemovesOnlyWhatThisAttemptRegistered: + + @pytest.mark.asyncio + async def test_a_manual_another_client_registered_during_the_attempt_survives_the_rollback(self, isolated_registries): + # While attempt A is registering its batch, client B (same shared + # repository) registers "contested". A's own registration of + # "contested" then fails as a duplicate; A fails outright on the bad + # manual. The rollback must remove what A registered -- and not B's + # "contested", which merely appeared after A started. + repo = InMemToolRepository() + client_b = await UtcpClient.create(config=UtcpClientConfig(tool_repository=repo)) + + class InterleavingProtocol(RecordingProtocol): + async def register_manual(self, caller, manual_call_template): + if manual_call_template.name == "trigger": + await client_b.register_manual(http_manual("contested")) + return await super().register_manual(caller, manual_call_template) + + CommunicationProtocol.communication_protocols["http"] = InterleavingProtocol() + + with pytest.raises(UtcpVariableNotFound): + await UtcpClient.create(config=UtcpClientConfig( + tool_repository=repo, + manual_call_templates=[http_manual("trigger"), http_manual("contested"), unresolvable_http_manual("bad_manual")], + )) + + assert await repo.get_manual("trigger") is None # A's own registration: rolled back + assert await repo.get_manual("contested") is not None # B's: untouched + await client_b.close() + + +class TestCancellationDuringRollbackStillClosesOwnedProtocols: + + @pytest.mark.asyncio + async def test_cancel_while_deregistering_closes_the_protocols_and_propagates_the_cancellation(self, isolated_registries): + # The rollback's deregistration blocks; the caller cancels create() + # while it is blocked. The owned protocol must still be closed, and + # the caller must see the cancellation. + deregister_started = asyncio.Event() + release_deregister = asyncio.Event() + + class BlockingDeregisterProtocol(RecordingProtocol): + async def deregister_manual(self, caller, manual_call_template): + deregister_started.set() + await release_deregister.wait() + + made: List[RecordingProtocol] = [] + + def factory() -> RecordingProtocol: + protocol = BlockingDeregisterProtocol() + made.append(protocol) + return protocol + + CommunicationProtocol.communication_protocol_factories["http"] = factory + + creating = asyncio.create_task(UtcpClient.create(config={"manual_call_templates": [ + {"name": "good_manual", "call_template_type": "http", "url": "https://example.test/utcp", "http_method": "POST"}, + {"name": "bad_manual", "call_template_type": "http", "url": "https://example.test/${NOWHERE_TO_BE_FOUND}", "http_method": "POST"}, + ]})) + await asyncio.wait_for(deregister_started.wait(), timeout=5) + creating.cancel() + + with pytest.raises(asyncio.CancelledError): + await creating + + assert len(made) == 1 + assert made[0].closed == 1 diff --git a/core/tests/client/test_utcp_client.py b/core/tests/client/test_utcp_client.py index 934bebf..9a5d951 100644 --- a/core/tests/client/test_utcp_client.py +++ b/core/tests/client/test_utcp_client.py @@ -12,6 +12,7 @@ from utcp.implementations.utcp_client_implementation import UtcpClientImplementation from utcp.interfaces.communication_protocol import CommunicationProtocol from utcp.utcp_client import UtcpClient +from utcp.plugins.plugin_loader import ensure_plugins_initialized from utcp.data.utcp_client_config import UtcpClientConfig from utcp.exceptions import UtcpVariableNotFound, UtcpSerializerValidationError from utcp.interfaces.concurrent_tool_repository import ConcurrentToolRepository @@ -187,7 +188,12 @@ async def sample_tools(): @pytest.fixture def isolated_communication_protocols(monkeypatch): - """Isolates the CommunicationProtocol registry for each test.""" + """Isolates the CommunicationProtocol registry for each test. + + Plugins are loaded first so that their registrations land in the original + dict (restored after the test), not in the throwaway one. + """ + ensure_plugins_initialized() monkeypatch.setattr(CommunicationProtocol, "communication_protocols", {}) diff --git a/scripts/extract_required_docs.py b/scripts/extract_required_docs.py index 92cfa3c..debff6e 100644 --- a/scripts/extract_required_docs.py +++ b/scripts/extract_required_docs.py @@ -546,15 +546,30 @@ def add_cross_references_post_generation(self, text: str, current_output_file: s class_anchor = re.sub(r'[^\w\-_]', '-', class_name.lower()).strip('-') link = f"[{class_name}](./{relative_path_str}#{class_anchor})" - # Don't replace matches that are in code blocks or inline code spans - lines = modified_text.split('\n') + # Don't replace matches that are in code blocks or inline code + # spans. Fenced blocks are delimited per line; everything + # between fences is handed to the inline-span scanner as ONE + # segment, so a span that crosses a newline stays a span. + segments: List[str] = [] + prose: List[str] = [] in_code_block = False - for i, line in enumerate(lines): + + def flush_prose() -> None: + if prose: + segments.append(self._sub_outside_inline_code(pattern, link, '\n'.join(prose))) + prose.clear() + + for line in modified_text.split('\n'): if line.strip().startswith('```'): + flush_prose() in_code_block = not in_code_block - elif not in_code_block: - lines[i] = self._sub_outside_inline_code(pattern, link, line) - modified_text = '\n'.join(lines) + segments.append(line) + elif in_code_block: + segments.append(line) + else: + prose.append(line) + flush_prose() + modified_text = '\n'.join(segments) return modified_text