http-api: T7090: Implement background configure operations for REST API

Large config commits (`service config-sync`) can block the REST API request
path and sometimes must be deferred (e.g., when changing `service https`).

This commit introduces an in-memory background operation manager
that queues (FIFO) full configure operations (commands + commit/commit-confirm)
as single jobs, tracks status/result, and exposes active operations
via `/retrieve/background-operations`.

4 files changed, 463 insertions, 45 deletions

diff --git a/smoketest/scripts/cli/test_service_https.py b/smoketest/scripts/cli/test_service_https.py
index 7fee7158d..4585bb0bd 100755
--- a/smoketest/scripts/cli/test_service_https.py
+++ b/smoketest/scripts/cli/test_service_https.py
@@ -17,6 +17,7 @@
 import unittest
 import json
 import psutil
+import time
 
 from requests import request
 from urllib3.exceptions import InsecureRequestWarning
@@ -35,6 +36,9 @@ from vyos.configsession import ConfigSessionError
 base_path = ['service', 'https']
 pki_base = ['pki']
 
+address = '127.0.0.1'
+key = 'VyOS-key'
+
 cert_data = """
 MIICFDCCAbugAwIBAgIUfMbIsB/ozMXijYgUYG80T1ry+mcwCgYIKoZIzj0EAwIw
 WTELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv
@@ -116,6 +120,40 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
         # always forward to base class
         super().tearDown()
 
+    def _api_get_background_operations(self):
+        url = f'https://{address}/retrieve/background-operations'
+        r = request('POST', url, verify=False, json={'key': key})
+        self.assertEqual(r.status_code, 200)
+        body = r.json()
+        self.assertTrue(body.get('success'))
+        ops = body.get('data', {}).get('operations', [])
+        return ops
+
+    def _wait_no_active_operations(self, timeout: int = 30):
+        # wait until no queued/running operations remain
+        deadline = time.time() + timeout
+        statuses = ('queued', 'running')
+
+        while time.time() < deadline:
+            ops = self._api_get_background_operations()
+            ops = [op for op in ops if op.get('status') in statuses]
+            if not ops:
+                return
+            sleep(0.25)
+        self.fail('Timeout waiting for background operations to finish')
+
+    def assertBackgroundOpResponseIsOk(self, response):
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertTrue(body.get('success'))
+
+        data = body.get('data', {})
+        self.assertIsInstance(data, dict)
+        self.assertIn('operation', data)
+        op = data['operation']
+        self.assertIn('op_id', op)
+        self.assertIn('status', op)
+
     def test_listen_address(self):
         test_prefix = ['192.0.2.1/26', '2001:db8:1::ffff/64']
         test_addr = [ i.split('/')[0] for i in test_prefix ]
@@ -356,8 +394,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
     @ignore_warning(InsecureRequestWarning)
     def test_api_add_delete(self):
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/retrieve'
         payload = {'data': '{"op": "showConfig", "path": []}', 'key': f'{key}'}
         headers = {}
@@ -387,8 +423,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
     @ignore_warning(InsecureRequestWarning)
     def test_api_show(self):
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/show'
         headers = {}
 
@@ -405,8 +439,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
     @ignore_warning(InsecureRequestWarning)
     def test_api_generate(self):
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/generate'
         headers = {}
 
@@ -423,8 +455,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
     @ignore_warning(InsecureRequestWarning)
     def test_api_configure(self):
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/configure'
         headers = {}
         conf_interface = 'dum0'
@@ -449,8 +479,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
     @ignore_warning(InsecureRequestWarning)
     def test_api_config_file(self):
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/config-file'
         headers = {}
 
@@ -467,8 +495,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
     @ignore_warning(InsecureRequestWarning)
     def test_api_reset(self):
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/reset'
         headers = {}
 
@@ -485,8 +511,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
     @ignore_warning(InsecureRequestWarning)
     def test_api_image(self):
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/image'
         headers = {}
 
@@ -528,8 +552,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
     @ignore_warning(InsecureRequestWarning)
     def test_api_config_file_load_http(self):
         # Test load config from HTTP URL
-        address = '127.0.0.1'
-        key = 'VyOS-key'
         url = f'https://{address}/config-file'
         url_config = f'https://{address}/configure'
         headers = {}
@@ -572,6 +594,125 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
         call(f'sudo rm -f {nginx_tmp_site}')
         call('sudo systemctl reload nginx')
 
+    @ignore_warning(InsecureRequestWarning)
+    def test_api_configure_background(self):
+        url = f'https://{address}/configure'
+        conf_interface = 'dum8'
+        conf_address = '192.0.2.88/32'
+
+        # Enable REST API
+        self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key])
+        self.cli_set(base_path + ['api', 'rest'])
+        self.cli_commit()
+
+        payload_path = [
+            'interfaces',
+            'dummy',
+            conf_interface,
+            'address',
+        ]
+        params = {'in_background': True}
+        payload = {
+            'data': json.dumps(
+                {'op': 'set', 'path': payload_path, 'value': conf_address}
+            ),
+            'key': key,
+        }
+
+        r = request('POST', url, verify=False, params=params, data=payload)
+        self.assertBackgroundOpResponseIsOk(r)
+        body = r.json()
+        op = body.get('data', {}).get('operation', [])
+
+        # Operation should appear as active shortly
+        ops = self._api_get_background_operations()
+        self.assertTrue(
+            any(o.get('op_id') == op.get('op_id') for o in ops),
+            'Queued operation is not visible in `/retrieve/background-operations`',
+        )
+
+        # Wait until done
+        self._wait_no_active_operations()
+
+        # Verify config applied (using CLI show)
+        self.assertIn(conf_address, self.op_mode(['show', 'configuration', 'commands']))
+
+    @ignore_warning(InsecureRequestWarning)
+    def test_api_configure_section_background(self):
+        url = f'https://{address}/configure-section'
+        conf_interface = 'dum9'
+        conf_address = '192.0.2.99/32'
+
+        # Enable REST API
+        self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key])
+        self.cli_set(base_path + ['api', 'rest'])
+        self.cli_commit()
+
+        # Configure-section payload: set a full section
+        # example: set interfaces dummy dum8 address 192.0.2.99/32
+        payload = {
+            'data': json.dumps(
+                {
+                    'op': 'set',
+                    'path': ['interfaces', 'dummy', conf_interface],
+                    'section': {
+                        'address': [conf_address],
+                    },
+                }
+            ),
+            'key': key,
+        }
+        params = {'in_background': True}
+
+        r = request('POST', url, verify=False, params=params, data=payload)
+        self.assertBackgroundOpResponseIsOk(r)
+
+        # Wait until done
+        self._wait_no_active_operations()
+
+        # Verify section applied
+        self.assertIn(conf_address, self.op_mode(['show', 'configuration', 'commands']))
+
+    @ignore_warning(InsecureRequestWarning)
+    def test_api_configure_background_ops_over_max(self):
+        max_ops = 128
+
+        # Enable REST API
+        self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key])
+        self.cli_set(base_path + ['api', 'rest'])
+        self.cli_commit()
+
+        op_ids = []
+        params = {'in_background': True}
+        url = f'https://{address}/configure'
+
+        # Create many non-existent configurations to fill the queue.
+        for i in range(max_ops + 5):
+            config_name = f'invalid-test-option-{i}'
+            payload_path = ['system', config_name]
+            payload = {
+                'data': json.dumps(
+                    {'op': 'set', 'path': payload_path, 'value': config_name}
+                ),
+                'key': key,
+            }
+
+            with self.subTest(payload_path=payload_path):
+                r = request('POST', url, verify=False, params=params, data=payload)
+                self.assertBackgroundOpResponseIsOk(r)
+
+            body = r.json()
+            op = body.get('data', {}).get('operation', [])
+            op_ids.append(op)
+
+        # Wait for queue to drain
+        self._wait_no_active_operations(timeout=120)
+
+        # Verify pruning: oldest `op_id` should be absent, and count should be <= `max_ops`
+        ops = self._api_get_background_operations()
+        self.assertLessEqual(len(ops), max_ops)
+        self.assertFalse(any(o.get('op_id') == op_ids[0] for o in ops))
+
 
 if __name__ == '__main__':
     unittest.main(verbosity=5)
diff --git a/src/helpers/vyos_config_sync.py b/src/helpers/vyos_config_sync.py
index 80bfb6d17..f0650c5fc 100755
--- a/src/helpers/vyos_config_sync.py
+++ b/src/helpers/vyos_config_sync.py
@@ -40,9 +40,12 @@ logger.name = os.path.basename(__file__)
 API_HEADERS = {'Content-Type': 'application/json'}
 
 
-def post_request(url: str,
-                 data: str,
-                 headers: Dict[str, str]) -> requests.Response:
+def post_request(
+    url: str,
+    data: str,
+    params: Dict[str, Any],
+    headers: Dict[str, str],
+) -> requests.Response:
     """Sends a POST request to the specified URL
 
     Args:
@@ -54,11 +57,14 @@ def post_request(url: str,
         requests.Response: The response object representing the server's response to the request
     """
 
-    response = requests.post(url,
-                             data=data,
-                             headers=headers,
-                             verify=False,
-                             timeout=timeout)
+    response = requests.post(
+        url,
+        data=data,
+        params=params,
+        headers=headers,
+        verify=False,
+        timeout=timeout,
+    )
     return response
 
 
@@ -116,6 +122,10 @@ def set_remote_config(
     urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
     url = f'https://{address}:{port}/configure-section'
+    params = {
+        # Ask the remote API to perform the configure and commit workflow asynchronously
+        'in_background': True,
+    }
     data = json.dumps({
         'op': op,
         'mask': mask,
@@ -124,7 +134,7 @@ def set_remote_config(
     })
 
     try:
-        config = post_request(url, data, headers)
+        config = post_request(url, data, params, headers)
         return config.json()
     except requests.exceptions.RequestException as e:
         print(f"An error occurred: {e}")
diff --git a/src/services/api/background.py b/src/services/api/background.py
new file mode 100644
index 000000000..2b25af307
--- /dev/null
+++ b/src/services/api/background.py
@@ -0,0 +1,179 @@
+# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this library.  If not, see <http://www.gnu.org/licenses/>.
+
+import time
+import functools
+from collections import deque
+from enum import Enum
+from threading import Lock
+from typing import Any
+from typing import Callable
+from typing import Optional
+from uuid import uuid4
+
+from fastapi import BackgroundTasks
+from pydantic import BaseModel
+from pydantic import StrictStr
+from pydantic import StrictInt
+
+
+def _ts():
+    """Return current Unix timestamp (seconds since epoch)"""
+    return int(time.time())
+
+
+class BackgroundOpStatus(str, Enum):
+    queued = 'queued'
+    running = 'running'
+    succeeded = 'succeeded'
+    failed = 'failed'
+
+    @property
+    def is_completed(self):
+        """True if the operation is in a terminal state (succeeded/failed)"""
+        return self in (BackgroundOpStatus.succeeded, BackgroundOpStatus.failed)
+
+
+class BackgroundOpRecord(BaseModel):
+    """Metadata and outcome for a single background operation"""
+
+    op_id: StrictStr
+    created_at: StrictInt
+    started_at: Optional[StrictInt] = None
+    finished_at: Optional[StrictInt] = None
+    status: BackgroundOpStatus = BackgroundOpStatus.queued
+    result: Optional[Any] = None
+    error: Optional[StrictStr] = None
+
+
+class BackgroundOpError(Exception):
+    """Raised when a background operation cannot be enqueued/executed"""
+
+    pass
+
+
+class BackgroundOpManager:
+    """
+    In-memory FIFO operation queue.
+
+    Uses BackgroundTasks to schedule a `drain()` call after the response,
+    so `enqueue()` is fast and non-blocking for the client.
+    """
+
+    DEFAULT_MAX_QUEUE_SIZE = 128
+
+    def __init__(self, max_queue_size: int = DEFAULT_MAX_QUEUE_SIZE):
+        # max number of queued (pending) operations allowed at a time
+        self._max_queue_size = max_queue_size
+
+        # FIFO queue of operation IDs waiting to be executed
+        self._queue = deque()
+        self._jobs = {}
+        self._workers = {}
+
+        # protects _queue/_jobs/_workers/_drain_scheduled from concurrent access
+        self._mx = Lock()
+
+        # whether a drain task has already been scheduled via BackgroundTasks
+        self._drain_scheduled = False
+
+    def enqueue(
+        self,
+        background_tasks: BackgroundTasks,
+        func: Callable,
+        *args,
+        **kwargs,
+    ) -> BackgroundOpRecord:
+        """Enqueue a function for background execution and return its record"""
+
+        assert isinstance(background_tasks, BackgroundTasks)
+        assert callable(func), '`func` argument should be function or lambda'
+
+        with self._mx:
+            if len(self._queue) >= self._max_queue_size:
+                raise BackgroundOpError(
+                    f'Background operation queue is full ({self._max_queue_size})'
+                )
+
+            op_id = str(uuid4())
+            record = BackgroundOpRecord(op_id=op_id, created_at=_ts())
+
+            self._jobs[op_id] = record
+            # store the callable for later execution (outside the lock)
+            self._workers[op_id] = functools.partial(func, *args, **kwargs)
+            self._queue.append(op_id)
+
+            if not self._drain_scheduled:
+                # schedule a single drain() call after the current response
+                background_tasks.add_task(self.drain)
+                self._drain_scheduled = True
+
+            # Best-effort pruning: keep history bounded by dropping oldest completed records
+            if len(self._jobs) > self._max_queue_size:
+                oldest = min(self._jobs.values(), key=lambda record: record.created_at)
+                if oldest.status.is_completed:
+                    del self._jobs[oldest.op_id]
+
+        return record
+
+    def drain(self):
+        """Run queued operations sequentially until the queue is empty"""
+
+        while True:
+            with self._mx:
+                if not self._queue:
+                    # allow future enqueue() calls to schedule the next drain()
+                    self._drain_scheduled = False
+                    return
+
+                op_id = self._queue.popleft()
+                record = self._jobs[op_id]
+                func = self._workers.pop(op_id)
+
+                record.status = BackgroundOpStatus.running
+                record.started_at = _ts()
+
+            # execute outside the lock to avoid blocking enqueues/status reads
+            result = error = status = None
+            try:
+                result = func()
+            except Exception as e:  # noqa: BLE001
+                status = BackgroundOpStatus.failed
+                error = str(e)
+            else:
+                status = BackgroundOpStatus.succeeded
+
+            with self._mx:
+                record.result = result
+                record.error = error
+                record.status = status
+                record.finished_at = _ts()
+
+    def get_record(self, op_id: str) -> BackgroundOpRecord | None:
+        """Return a deep copy of a single record"""
+
+        with self._mx:
+            record = self._jobs.get(op_id)
+            return record.copy(deep=True) if record else None
+
+    def get_records(self) -> list:
+        """Return deep copies of all records, sorted oldest-first by created_at"""
+
+        with self._mx:
+            records = [record.copy(deep=True) for record in self._jobs.values()]
+
+        # stable-ish ordering (oldest first)
+        records.sort(key=lambda record: record.created_at)
+        return records
diff --git a/src/services/api/rest/routers.py b/src/services/api/rest/routers.py
index 438b3a23e..c8102c980 100644
--- a/src/services/api/rest/routers.py
+++ b/src/services/api/rest/routers.py
@@ -28,6 +28,7 @@ from typing import Callable
 from typing import TYPE_CHECKING
 
 from fastapi import Depends
+from fastapi import Query
 from fastapi import Request
 from fastapi import Response
 from fastapi import HTTPException
@@ -46,6 +47,8 @@ from vyos.configtree import ConfigTree
 from vyos.configdiff import get_config_diff
 from vyos.configsession import ConfigSessionError
 
+from ..background import BackgroundOpManager
+from ..background import BackgroundOpError
 from ..session import SessionState
 from .models import success
 from .models import error
@@ -230,7 +233,7 @@ class MultipartRequest(Request):
                                 400,
                                 f"Malformed command '{0}': 'path' field must be a list of strings",
                             )
-                    if endpoint in ('/configure'):
+                    if endpoint in ('/configure',):
                         if not c['path']:
                             self.form_err = (
                                 400,
@@ -241,7 +244,7 @@ class MultipartRequest(Request):
                                 400,
                                 f"Malformed command '{c}': 'value' field must be a string",
                             )
-                    if endpoint in ('/configure-section'):
+                    if endpoint in ('/configure-section',):
                         if 'section' not in c and 'config' not in c:
                             self.form_err = (
                                 400,
@@ -293,6 +296,10 @@ router = APIRouter(
 self_ref_msg = 'Requested HTTP API server configuration change; commit will be called in the background'
 
 
+# Global background-op manager used by the REST API to run long config commits after the response
+background_op_manager = BackgroundOpManager()
+
+
 def call_commit(s: SessionState):
     try:
         s.session.commit()
@@ -342,7 +349,7 @@ def run_commit_confirm(s: SessionState):
         del env['IN_COMMIT_CONFIRM']
 
 
-async def _configure_op(
+def _execute_configure_op(
     data: Union[
         ConfirmModel,
         ConfigureModel,
@@ -351,12 +358,15 @@ async def _configure_op(
         ConfigSectionListModel,
         ConfigSectionTreeModel,
     ],
-    _request: Request,
-    background_tasks: BackgroundTasks,
+    background_tasks: BackgroundTasks | None = None,
 ):
     # pylint: disable=too-many-branches,too-many-locals,too-many-nested-blocks,too-many-statements
     # pylint: disable=consider-using-with
 
+    # True when invoked by the background operation
+    # runner (no FastAPI BackgroundTasks context passed in)
+    is_background_job = background_tasks is None
+
     state = SessionState()
     session = state.session
     env = session.get_session_env()
@@ -385,6 +395,8 @@ async def _configure_op(
     try:
         for c in data:
             op = c.op
+            op_error = ConfigSessionError(f"'{op}' is not a valid operation")
+
             if not isinstance(c, (ConfirmModel, BaseConfigSectionTreeModel)):
                 path = c.path
 
@@ -392,7 +404,7 @@ async def _configure_op(
                 if op == 'confirm':
                     msg = session.confirm()
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
 
             elif isinstance(c, BaseConfigureModel):
                 if c.value:
@@ -422,7 +434,7 @@ async def _configure_op(
                 elif op == 'comment':
                     session.comment(path, value=value)
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
 
             elif isinstance(c, BaseConfigSectionModel):
                 if op == 'set':
@@ -430,7 +442,7 @@ async def _configure_op(
                 elif op == 'load':
                     session.load_section(path, section)
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
 
             elif isinstance(c, BaseConfigSectionTreeModel):
                 if op == 'set':
@@ -438,7 +450,7 @@ async def _configure_op(
                 elif op == 'load':
                     session.load_section_tree(mask, config)
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
         # end for
 
         config = Config(session_env=env)
@@ -448,20 +460,29 @@ async def _configure_op(
 
         if not d.is_node_changed(['service', 'https']):
             if confirm_time:
-                out, err = await run_in_threadpool(run_commit_confirm, state)
+                out, err = run_commit_confirm(state)
                 if err:
                     raise err
                 msg = msg + out if msg else out
             else:
-                out, err = await run_in_threadpool(run_commit, state)
+                out, err = run_commit(state)
                 if err:
                     raise err
                 msg = msg + out if msg else out
         else:
-            if confirm_time:
-                background_tasks.add_task(call_commit_confirm, state)
+            if is_background_job:
+                # If already running as a background job, commit synchronously here
+                if confirm_time:
+                    call_commit_confirm(state)
+                else:
+                    call_commit(state)
             else:
-                background_tasks.add_task(call_commit, state)
+                # Otherwise schedule the commit to run after the HTTP response
+                if confirm_time:
+                    background_tasks.add_task(call_commit_confirm, state)
+                else:
+                    background_tasks.add_task(call_commit, state)
+
             out = self_ref_msg
             msg = msg + out if msg else out
 
@@ -484,12 +505,62 @@ async def _configure_op(
             del env['IN_COMMIT_CONFIRM']
         lock.release()
 
+    # Background jobs return raw success text or raise on failure;
+    # the API wrapper formats HTTP responses and returns it
+    if is_background_job:
+        if status == 200:
+            return msg
+        else:
+            raise RuntimeError(error_msg)
+
     if status != 200:
         return error(status, error_msg)
 
     return success(msg)
 
 
+async def _configure_op(
+    data: Union[
+        ConfirmModel,
+        ConfigureModel,
+        ConfigureListModel,
+        ConfigSectionModel,
+        ConfigSectionListModel,
+        ConfigSectionTreeModel,
+    ],
+    background_tasks: BackgroundTasks,
+    in_background: bool = False,
+):
+    """
+    API wrapper for configure operations.
+
+    If `in_background=True`: enqueue the whole configure
+    workflow and return an operation record immediately.
+    Otherwise: run the configure workflow in a threadpool
+    and return the normal API response.
+    """
+
+    if in_background:
+        try:
+            # Enqueue and return an operation handle that
+            # can be polled via `/retrieve/background-operations`
+            record = background_op_manager.enqueue(
+                background_tasks,
+                _execute_configure_op,
+                data,
+            )
+        except BackgroundOpError as e:
+            return error(500, str(e))
+
+        return success({'operation': record.model_dump()})
+
+    return await run_in_threadpool(
+        _execute_configure_op,
+        data,
+        background_tasks=background_tasks,
+    )
+
+
 def create_path_import_pki_no_prompt(path):
     correct_paths = ['ca', 'certificate', 'key-pair']
     if path[1] not in correct_paths:
@@ -504,10 +575,9 @@ async def configure_op(
     data: Union[ConfigureModel, ConfigureListModel, ConfirmModel],
     request: Request,
     background_tasks: BackgroundTasks,
+    in_background: bool = Query(False),
 ):
-    out = await _configure_op(data, request, background_tasks)
-
-    return out
+    return await _configure_op(data, background_tasks, in_background)
 
 
 @router.post('/configure-section')
@@ -515,10 +585,9 @@ async def configure_section_op(
     data: Union[ConfigSectionModel, ConfigSectionListModel, ConfigSectionTreeModel],
     request: Request,
     background_tasks: BackgroundTasks,
+    in_background: bool = Query(False),
 ):
-    out = await _configure_op(data, request, background_tasks)
-
-    return out
+    return await _configure_op(data, background_tasks, in_background)
 
 
 @router.post('/retrieve')
@@ -565,6 +634,25 @@ async def retrieve_op(data: RetrieveModel):
     return success(res)
 
 
+@router.post('/retrieve/background-operations')
+async def retrieve_background_operations(
+    op_id: str = Query(None),
+):
+    if op_id:
+        # Return only that record
+        record = background_op_manager.get_record(op_id)
+        records = [record] if record else []
+    else:
+        # Return the full in-memory operation history (oldest first)
+        records = background_op_manager.get_records()
+
+    result = {
+        'operations': [record.model_dump() for record in records],
+    }
+
+    return success(result)
+
+
 @router.post('/config-file')
 async def config_file_op(data: ConfigFileModel, background_tasks: BackgroundTasks):
     state = SessionState()