Merge branch 'aiidateam:main' into fix/scp-glob-escaping

Author: stillfast

.docker/aiida-core-base/s6-assets/init/aiida-prepare.sh

@@ -18,8 +18,11 @@ verdi config set warnings.development_version False
 # If the environment variable `SETUP_DEFAULT_AIIDA_PROFILE` is not set, set it to `true`.
 if [[ ${SETUP_DEFAULT_AIIDA_PROFILE:-true} == true ]] && ! verdi profile show ${AIIDA_PROFILE_NAME} &> /dev/null; then
 
-    # Create AiiDA profile.
-    verdi presto \
+    # Create AiiDA profile. Set AIIDA_NO_DAEMON_AUTOSTART because the
+    # Docker container manages the daemon lifecycle separately via
+    # s6-overlay (aiida-daemon-start runs after this script and
+    # run-before-daemon-start).
+    AIIDA_NO_DAEMON_AUTOSTART=1 verdi presto \
         --verbosity info \
         --profile-name "${AIIDA_PROFILE_NAME:-default}" \
         --email "${AIIDA_USER_EMAIL:-aiida@localhost}" \

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## Unreleased
+
+### Behavior changes
+
+**`verdi presto`: auto-starts the daemon** ([#7351](https://github.com/aiidateam/aiida-core/pull/7351))
+
+`verdi presto` now starts the daemon automatically after creating the profile, when a broker is configured.
+Scripts that previously ran `verdi daemon start` after `verdi presto` continue to work; the daemon-start invocation is idempotent.
+
 ## v2.8.0 - 2026-03-16
 
 This release brings important improvements to the `BaseRestartWorkChain`, the engine, stashing, typing coverage, and dependency updates.

docs/source/internals/broker.rst

@@ -137,8 +137,10 @@ Message types
      - worker → broker
      - Negative acknowledgment. Broker requeues the task for redelivery.
    * - ``TASK_RESPONSE``
-     - worker → broker → client
-     - Return the result of a completed task. Broker forwards to original sender.
+     - broker → client
+     - Immediate acknowledgment that the task was persisted to the queue
+       (matches RabbitMQ publisher-confirm semantics).
+       The worker's eventual result is not forwarded back to the sender.
    * - ``RPC``
      - client → broker → recipient
      - Remote procedure call to a named recipient. Broker routes by subscriber ID.
@@ -173,6 +175,7 @@ The protocol maps AMQP concepts to ZMQ message types:
 
     AMQP concept              ZMQ broker equivalent
     ────────────────────────  ──────────────────────────────────
+    publisher confirm         TASK_RESPONSE (immediate broker ack)
     basic.ack                 TASK_ACK
     basic.nack                TASK_NACK
     consumer with prefetch    TASK dispatch to available workers
@@ -209,24 +212,27 @@ Message flow: task submission
           │  task_send(task)            │                      │
           │──── TASK ──────────────────────────────────────────▶│
           │                             │          push to PersistentQueue
+          │◀──── TASK_RESPONSE (ack) ──────────────────────────│  immediate
           │                             │                      │
           │                             │◀─── TASK ────────────│  dispatch
           │                             │                      │
           │                             │  (process runs...)   │
           │                             │                      │
           │                             │──── TASK_ACK ───────▶│  ack: remove from queue
-          │                             │──── TASK_RESPONSE ──▶│
-          │                             │                      │
-          │◀────────────── TASK_RESPONSE (forwarded) ──────────│
           │                             │                      │
 
 Key details:
 
+- The broker sends an **immediate** ``TASK_RESPONSE`` back to the sender as soon as the task
+  is persisted to the queue on disk.  This matches RabbitMQ's publisher-confirm semantics:
+  the caller's ``Future`` resolves without waiting for a worker to process the task.
+  Without this, ``task_send`` would block until ``broker.task_timeout`` when no workers are connected
+  (e.g. during ``verdi process repair``).
 - The broker persists the task to disk **before** dispatching it.
   If the broker crashes, tasks are recovered from disk on restart.
 - Workers delay the ACK until the task Future resolves (see :ref:`deferred ACK <internal_architecture:broker:deferred_ack>` below).
   If a worker dies before ACK'ing, the broker detects the disconnect (via ZMTP heartbeats + PING probing) and requeues the task.
-- If ``no_reply=True``, no TASK_RESPONSE is sent.
+- If ``no_reply=True``, no ``TASK_RESPONSE`` is sent (fire-and-forget, used by ``submit()``).
 
 
 Task dispatch strategy

docs/source/reference/command_line.rst

@@ -319,6 +319,7 @@ Below is a list with all available subcommands.
       * Create a default user for the profile (email can be configured through the `--email` option)
       * Set up the localhost as a `Computer` and configure it
       * Set a number of configuration options with sensible defaults
+      * Start the daemon (unless `--no-broker` is specified)
 
       By default the command creates a profile that uses SQLite for the database. For the
       message broker, it automatically checks for RabbitMQ running on localhost. If found, it

pyproject.toml

@@ -22,6 +22,7 @@ classifiers = [
   'Programming Language :: Python :: 3.11',
   'Programming Language :: Python :: 3.12',
   'Programming Language :: Python :: 3.13',
+  'Programming Language :: Python :: 3.14',
   'Topic :: Scientific/Engineering'
 ]
 # NOTE: When updating versions of some packages, such as requests or paramiko,

src/aiida/brokers/rabbitmq/defaults.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
 import os
 import typing as t
+from contextlib import contextmanager, suppress
 
 from aiida.common.extendeddicts import AttributeDict
 from aiida.common.log import AIIDA_LOGGER
@@ -27,6 +30,49 @@
 )
 
 
+@contextmanager
+def _suppress_rabbitmq_probe_logging() -> t.Iterator[None]:
+    """Suppress expected RabbitMQ probe logs for missing brokers."""
+    logger_names = ('aiormq.connection', 'aio_pika.robust_connection')
+    levels = {name: logging.getLogger(name).level for name in logger_names}
+
+    try:
+        for name in logger_names:
+            logging.getLogger(name).setLevel(logging.CRITICAL + 1)
+        yield
+    finally:
+        for name, level in levels.items():
+            logging.getLogger(name).setLevel(level)
+
+
+def _probe_rabbitmq_connection(connection_params: dict[str, t.Any]) -> None:
+    """Validate RabbitMQ connection parameters in an isolated event loop."""
+    from aio_pika.connection import make_url
+    from aio_pika.robust_connection import RobustConnection
+
+    async def connect() -> None:
+        connection = RobustConnection(
+            make_url(
+                host=connection_params['host'],
+                port=connection_params['port'],
+                login=connection_params['username'],
+                password=connection_params['password'],
+                virtualhost=connection_params['virtual_host'],
+                ssl=connection_params['protocol'] == 'amqps',
+            ),
+            loop=asyncio.get_running_loop(),
+        )
+
+        try:
+            await connection.connect()
+        finally:
+            with suppress(Exception):
+                await connection.close()
+
+    with _suppress_rabbitmq_probe_logging():
+        asyncio.run(connect())
+
+
 def detect_rabbitmq_config(
     protocol: str | None = None,
     username: str | None = None,
@@ -38,10 +84,8 @@ def detect_rabbitmq_config(
     """Try to connect to a RabbitMQ server with the default connection parameters.
 
     :raises ConnectionError: If the connection failed with the provided connection parameters
-    :returns: The connection parameters if the RabbitMQ server was successfully connected to, or ``None`` otherwise.
+    :returns: The connection parameters with keys prefixed with ``broker_``.
     """
-    from kiwipy.rmq.threadcomms import connect
-
     connection_params = {
         'protocol': protocol or os.getenv('AIIDA_BROKER_PROTOCOL', BROKER_DEFAULTS['protocol']),
         'username': username or os.getenv('AIIDA_BROKER_USERNAME', BROKER_DEFAULTS['username']),
@@ -54,9 +98,10 @@ def detect_rabbitmq_config(
     LOGGER.info(f'Attempting to connect to RabbitMQ with parameters: {connection_params}')
 
     try:
-        connect(connection_params=connection_params)
-    except ConnectionError:
-        raise ConnectionError(f'Failed to connect with following connection parameters: {connection_params}')
+        _probe_rabbitmq_connection(connection_params)
+    except Exception as exception:
+        msg = f'Failed to connect with following connection parameters: {connection_params}'
+        raise ConnectionError(msg) from exception
 
     # The profile configuration expects the keys of the broker config to be prefixed with ``broker_``.
     return {f'broker_{key}': value for key, value in connection_params.items()}

src/aiida/brokers/zmq/communicator.py

@@ -205,6 +205,15 @@ def _on_loop() -> None:
             if threading.current_thread() is not self._loop_thread:
                 self._loop_thread.join(timeout=LOOP_JOIN_TIMEOUT)
 
+                if self._loop_thread.is_alive():
+                    _LOGGER.warning('ZMQ loop thread did not join in time, terminating ZMQ context')
+                    # Force-terminate the ZMQ context to unblock the poll loop.
+                    # This causes recv_multipart to raise ZMQError, exiting the thread.
+                    if self._context is not None:
+                        self._context.term()
+                        self._context = None
+                    self._loop_thread.join(timeout=LOOP_JOIN_TIMEOUT)
+
         self._loop = None
         self._loop_thread = None
 

src/aiida/brokers/zmq/server.py

@@ -85,8 +85,7 @@ def __init__(
         # rpc_subscribers: identifier -> client_identity (bytes)
         self._rpc_subscribers: dict[str, bytes] = {}
 
-        # Pending responses: correlation_id -> (client_identity, timestamp)
-        self._pending_task_responses: dict[str, tuple[bytes, float]] = {}
+        # Pending RPC responses: correlation_id -> (client_identity, timestamp)
         self._pending_rpc_responses: dict[str, tuple[bytes, float]] = {}
 
         # Task-worker assignments: task_id -> worker_identity
@@ -283,6 +282,14 @@ def _handle_task(self, identity: bytes, msg: dict[str, Any]) -> None:
         """Handle incoming task message.
 
         Queue the task and try to dispatch to an available worker.
+
+        When the sender expects a reply (``no_reply=False``), an immediate
+        acknowledgment response is sent back as soon as the task is persisted
+        to disk.  This mirrors RabbitMQ's publisher-confirm behaviour: the
+        caller learns that the broker accepted the task without having to wait
+        for a worker to process it.  The worker's eventual result (if any) is
+        *not* forwarded back to the original sender — callers like
+        ``continue_process`` only need the confirmation, not the outcome.
         """
         task_id = msg['id']
         sender = msg.get('sender', '')
@@ -299,33 +306,27 @@ def _handle_task(self, identity: bytes, msg: dict[str, Any]) -> None:
         }
         self._task_queue.push(task_id, task_data)
 
-        # Track pending response if reply expected
+        # Send an immediate acknowledgment to the sender so its Future
+        # resolves without waiting for a worker (matches RabbitMQ semantics).
         if not no_reply:
-            self._pending_task_responses[task_id] = (identity, time.time())
+            response = {
+                'type': MessageType.TASK_RESPONSE.value,
+                'task_id': task_id,
+                'result': True,
+            }
+            self._send_to_client(identity, response)
 
         # Try to dispatch immediately
         self._dispatch_pending_tasks()
 
     def _handle_task_response(self, identity: bytes, msg: dict[str, Any]) -> None:
         """Handle task response from worker.
 
-        Route the response back to the original task sender.
+        Since the broker already sends an immediate acknowledgment when a task
+        is queued, the worker's result response is simply logged and discarded.
         """
-        task_id = msg.get('task_id')
-        if not task_id:
-            _LOGGER.warning('Task response missing task_id')
-            return
-
-        # Find original sender
-        pending = self._pending_task_responses.pop(task_id, None)
-        if not pending:
-            _LOGGER.warning('No pending response for task: %s', task_id)
-            return
-
-        original_sender, _ = pending
-
-        # Forward response to original sender
-        self._send_to_client(original_sender, msg)
+        task_id = msg.get('task_id', '?')
+        _LOGGER.debug('Received task response for %s (discarded — sender already acknowledged)', task_id)
 
     def _handle_task_ack(self, identity: bytes, msg: dict[str, Any]) -> None:
         """Handle task acknowledgment from worker."""
@@ -602,7 +603,6 @@ def get_status(self) -> dict[str, Any]:
             'task_subscribers': len(self._task_subscribers),
             'rpc_subscribers': len(self._rpc_subscribers),
             'available_workers': len(self._available_workers),
-            'pending_task_responses': len(self._pending_task_responses),
             'pending_rpc_responses': len(self._pending_rpc_responses),
         }
 

src/aiida/brokers/zmq/service.py

@@ -93,9 +93,6 @@ def start(self) -> None:
         self._write_status(self._server.get_status())
 
     def stop(self) -> None:
-        if not self._running:
-            return
-
         self._running = False
 
         if self._server:

src/aiida/cmdline/commands/cmd_presto.py

@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import os
 import pathlib
 import re
 import typing as t
@@ -183,6 +184,7 @@ def verdi_presto(
     * Create a default user for the profile (email can be configured through the `--email` option)
     * Set up the localhost as a `Computer` and configure it
     * Set a number of configuration options with sensible defaults
+    * Start the daemon (unless `--no-broker` is specified)
 
     By default the command creates a profile that uses SQLite for the database. For the message broker, it automatically
     checks for RabbitMQ running on localhost. If found, it configures RabbitMQ as the broker. Otherwise, it falls back
@@ -289,3 +291,20 @@ def verdi_presto(
     computer.set_default_mpiprocs_per_machine(1)
 
     echo.echo_success('Configured the localhost as a computer.')
+
+    # `AIIDA_NO_DAEMON_AUTOSTART` is an internal escape hatch for the Docker init script (see
+    # `.docker/aiida-core-base/s6-assets/init/aiida-prepare.sh`); deliberately undocumented in `--help`.
+    if broker_backend is not None and os.environ.get('AIIDA_NO_DAEMON_AUTOSTART') != '1':
+        from aiida.common.exceptions import ConfigurationError
+        from aiida.engine.daemon.client import DaemonException, get_daemon_client
+
+        echo.echo('Starting the daemon... ', nl=False)
+        try:
+            client = get_daemon_client(profile.name)
+            client.start_daemon()
+        except (DaemonException, ConfigurationError) as exception:
+            echo.echo('FAILED', fg=echo.COLORS['error'], bold=True)
+            echo.echo_warning(f'Failed to start the daemon: {exception}')
+            echo.echo_report('You can start it manually with `verdi daemon start`.')
+        else:
+            echo.echo('OK', fg=echo.COLORS['success'], bold=True)