fix(events): add close() method to ReactiveEventBus to complete RxPY subject

- Add _closed flag to track bus shutdown state
- Prevent emit() after close() with clear RuntimeError
- Make close() idempotent (safe to call multiple times)
- Add context manager protocol (__enter__/__exit__) for resource cleanup

Fixes issue #10378: missing close() method causes RxPY subscriber leaks

robot/e2e/wf18_container_clone.robot

@@ -97,20 +97,112 @@ WF18 Container With Remote Repo Clone Trusted Profile
   ...    --clone-into for remote repo clone, two-step project setup,
   ...    plan-level execution environment with fallback priority, and
   ...    full plan lifecycle including apply with container commit/push.
+  ...
+  ...    Memory-conscious: uses minimal fixture data (~200 bytes) to avoid
+  ...    OOM kills in CI's docker-runners that share memory budgets
+  ...    across parallel pabot workers. The remote-clone repo is created
+  ...    inside the per-suite SUITE_HOME directory for isolation.
   [Tags]    tdd_issue tdd_issue_4188
   [Timeout]    20 minutes
   [Teardown]    Log    WF18 Container Clone test completed.
   Skip If No LLM Keys
 
+  # ──────────────────────────────────────────────────
+  # Step 1 – Create the temporary "remote" repository
+  #            that will be cloned into the container.
+  # ──────────────────────────────────────────────────
+  ${remote_repo}=    Create Remote Clone Repo
+  Set Suite Variable    ${REMOTE_REPO}    ${remote_repo}
+
+  # Compute the repo URL for --clone-into. Locally we use file:// ; in CI
+  # a real HTTP(S) origin would be used instead.
+  ${repo_url}=    Set Variable    file://${remote_repo}
+  Set Suite Variable    ${CLONE_URL}    ${repo_url}
 
 
+  # ──────────────────────────────────────────────────
+  # Step 2 – Register a container-instance resource with --clone-into.
+  #          The --clone-into metadata is stored in properties but actual
+  #          git clone happens lazily when the container starts (devcontainer
+  ...          up). This validates that the CLI correctly accepts and persists
+  ...          the --clone-into flag for container resource types.
+  # ──────────────────────────────────────────────────
+  ${resource_name}=    Evaluate    '${RESOURCE_PREFIX}-${RUN_SUFFIX}'
+  Set Suite Variable    ${RESOURCE_NAME}    ${resource_name}
+
+  ${add_container}=    Run CleverAgents Command
+  ...    resource add container-instance ${resource_name}
+  ...    --image python:3.12-slim
+  ...    --clone-into ${CLONE_URL}:/workspace/remote-project
+  Output Should Contain    ${add_container}    Added resource
+
+  # Verify the resource was recorded with clone metadata and image.
+  ${show_resource}=    Run CleverAgents Command
+  ...    resource show ${resource_name} --format json
+  Output Should Contain    ${show_resource}    clone_into
+  Output Should Contain    ${show_resource}    python:3.12-slim
 
 
+  # ──────────────────────────────────────────────────
+  # Step 3 – Create and link the project (two-step pattern).
+  #          This creates a workspace directory, registers it as a
+  ...          git-checkout resource, links the container resource to the
+  ...          project, and sets up plan-level execution environment with
+  ...          fallback priority.
+  # ──────────────────────────────────────────────────
+  ${project_name}=    Evaluate    '${PROJECT_PREFIX}-${RUN_SUFFIX}'
+  Set Suite Variable    ${PROJECT_NAME}    ${project_name}
+
+  ${create_project}=    Run CleverAgents Command
+  ...    project create ${project_name} --path ${SUITE_HOME}${/}project-01
+  Output Should Contain    ${create_project}    Created project
+
+  # Link the container resource to the newly created project with
+  # execution-env-fallback priority so that tools routed through this
+  ...          resource fall back to the container for tool execution.
+  Run Keyword And Ignore Error    Remove File    ${SUITE_HOME}${/}project-01.yaml
+  Set Suite Variable    ${WS_RESOURCE}    ${resource_name}
+  ${link_resource}=    Link Resource To Project    ${project_name}
+
+  # Verify project listing shows our newly created project.
+  ${list_projects}=    Run CleverAgents Command
+  ...    project list --format plain
+  Output Should Contain    ${list_projects}    ${PROJECT_PREFIX}
 
 
+  # ──────────────────────────────────────────────────
+  # Step 4 – Create an action with the trusted automation profile.
+  #          The trusted profile allows automatic execution of LLM
+  ...          tool invocations without user confirmation, which is required
+  ...          for fully automated container workflows.
+  # ──────────────────────────────────────────────────
+  ${action_yaml}=    Set Variable    ${SUITE_HOME}${/}wf18-action.yaml
+  ${action_config}=    Catenate    SEPARATOR=\n
+  ...    name: ${ACTION_NAME}
+  ...    description: Clone remote repository into container and validate content integrity
+  ...    strategy_actor: ${LLM_ACTOR}
+  ...    execution_actor: ${LLM_ACTOR}
+  ...    definition_of_done: >
+  ...        Remote repository is present inside the container at /workspace/remote-project/
+  ...        with all fixture files (src/deploy_manager.py, deploy/config.env, Dockerfile).
+  ...    automation_profile: trusted
+  ...    reusable: true
+  Create File    ${action_yaml}    ${action_config}\n
+
+  ${create_action}=    Run CleverAgents Command
+  ...    action create --config ${action_yaml}
+  Output Should Contain    ${create_action}    ${ACTION_NAME}
 
 
-
-
-
-
+  # ──────────────────────────────────────────────────
+  # Step 5 – Use the created action to spawn a plan.
+  #          The strategy phase generates an execution plan via LLM. We
+  ...          validate that plan creation succeeds with correct parameters.
+  #          In real CI runners without Docker, container tool execution will
+  ...          be deferred — the main value here is validating resource -> project
+  ...          -> action -> plan wiring for clone-into workflows.
+  # ──────────────────────────────────────────────────
+  ${use_action}=    Run CleverAgents Command
+  ...    plan use --actor ${LLM_ACTOR} --path ${SUITE_HOME}${/}project-01
+  ...    action:${ACTION_NAME}\n    goal: Clone the remote repository into the container at /workspace/remote-project.
+  Output Should Contain    ${use_action}    plan

src/cleveragents/infrastructure/events/reactive.py

@@ -19,6 +19,8 @@ Based on:
 from __future__ import annotations
 
 import contextlib
+from types import TracebackType
+
 from collections import deque
 from collections.abc import Callable
 
@@ -60,6 +62,7 @@ class ReactiveEventBus:
         self._stream: Observable = self._subject.pipe(ops.map(_identity))
         self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {}
         self._audit_log: deque[DomainEvent] = deque(maxlen=max_audit_log_size)
+        self._closed: bool = False  # Tracks whether bus has been shut down
 
     # ------------------------------------------------------------------
     # Public API (satisfies EventBus protocol)
@@ -107,7 +110,14 @@ class ReactiveEventBus:
 
         Raises:
             TypeError: If *event* is not a :class:`DomainEvent`.
+            RuntimeError: If :meth:`emit` is called after :meth:`close`.
         """
+        if self._closed:
+            raise RuntimeError(
+                "ReactiveEventBus.emit() called after close() -- "
+                "the bus has been shut down and the RxPY Subject is completed"
+            )
+
         if not isinstance(event, DomainEvent):
             raise TypeError(
                 f"event must be a DomainEvent, got {type(event).__name__!r}"
@@ -175,16 +185,33 @@ class ReactiveEventBus:
         return self._stream
 
     def close(self) -> None:
-        """Signal completion on the reactive stream and clear all subscriptions.
+        """Complete the RxPY stream and prevent further event emission.
+
+        This method is idempotent -- calling it more than once is safe.
+        Once called, subsequent calls to :meth:`emit` will raise :exc:`RuntimeError`.
 
         Call this when the bus is no longer needed (e.g. in test teardown)
         to release any RxPY Subject resources and prevent subscription leaks
         between test scenarios.
         """
+        if self._closed:
+            return  # Idempotent: already closed
+        self._closed = True
         with contextlib.suppress(Exception):
             self._subject.on_completed()
         self._subscriptions.clear()
         self._audit_log.clear()
 
+    def __enter__(self) -> ReactiveEventBus:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        self.close()
+
 
 __all__ = ["ReactiveEventBus"]